mirror of
https://github.com/msoedov/agentic_security.git
synced 2026-10-01 04:09:36 +02:00
Fixed issue 137: ensuring http/https with :// is present
This commit is contained in:
1 parent
f9dce3f915
commit
29362aed30
1 file changed
+18
-5
@@ -1,6 +1,7 @@
|
|||||||
import base64
|
import base64
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
|
|
||||||
|
from urllib.parse import urlparse
|
||||||
import httpx
|
import httpx
|
||||||
from pydantic import BaseModel
|
from pydantic import BaseModel
|
||||||
|
|
||||||
@@ -14,7 +15,7 @@ class Modality(Enum):
|
|||||||
FILES = 3
|
FILES = 3
|
||||||
MIXED = 4
|
MIXED = 4
|
||||||
|
|
||||||
|
# encoding the image data fro
|
||||||
def encode_image_base64_by_url(url: str = "https://github.com/fluidicon.png") -> str:
|
def encode_image_base64_by_url(url: str = "https://github.com/fluidicon.png") -> str:
|
||||||
"""Encode image data to base64 from a URL"""
|
"""Encode image data to base64 from a URL"""
|
||||||
response = httpx.get(url)
|
response = httpx.get(url)
|
||||||
@@ -39,10 +40,11 @@ class LLMSpec(BaseModel):
|
|||||||
headers: dict
|
headers: dict
|
||||||
body: str
|
body: str
|
||||||
has_files: bool = False
|
has_files: bool = False
|
||||||
has_image: bool = False
|
has_image: bool = Fa
|
||||||
has_audio: bool = False
|
has_audio: bool = False
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
|
# class method6
|
||||||
def from_string(cls, http_spec: str):
|
def from_string(cls, http_spec: str):
|
||||||
try:
|
try:
|
||||||
return parse_http_spec(http_spec)
|
return parse_http_spec(http_spec)
|
||||||
@@ -78,6 +80,8 @@ class LLMSpec(BaseModel):
|
|||||||
if self.has_audio and not encoded_audio:
|
if self.has_audio and not encoded_audio:
|
||||||
raise ValueError("Audio is required for this request.")
|
raise ValueError("Audio is required for this request.")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
async def probe(
|
async def probe(
|
||||||
self, prompt: str, encoded_image: str = "", encoded_audio: str = "", files={}
|
self, prompt: str, encoded_image: str = "", encoded_audio: str = "", files={}
|
||||||
) -> httpx.Response:
|
) -> httpx.Response:
|
||||||
@@ -144,9 +148,7 @@ def parse_http_spec(http_spec: str) -> LLMSpec:
|
|||||||
"""Parses an HTTP specification string into a LLMSpec object.
|
"""Parses an HTTP specification string into a LLMSpec object.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
http_spec (str): A string representing an HTTP specification.
|
http_spec (str): A string representing an HTTP specification. Returns:
|
||||||
|
|
||||||
Returns:
|
|
||||||
LLMSpec: An object representing the parsed HTTP specification, with attributes for the method, URL, headers, and body.
|
LLMSpec: An object representing the parsed HTTP specification, with attributes for the method, URL, headers, and body.
|
||||||
"""
|
"""
|
||||||
from agentic_security.core.app import get_secrets
|
from agentic_security.core.app import get_secrets
|
||||||
@@ -159,6 +161,17 @@ def parse_http_spec(http_spec: str) -> LLMSpec:
|
|||||||
# Extract the method and URL from the first line
|
# Extract the method and URL from the first line
|
||||||
method, url = lines[0].split(" ")[0:2]
|
method, url = lines[0].split(" ")[0:2]
|
||||||
|
|
||||||
|
# Check url validity
|
||||||
|
|
||||||
|
valid_url= urlparse(url)
|
||||||
|
#
|
||||||
|
if valid_url.scheme not in ("http", "https") or not valid_url.netloc: # if missing the correct formatting ://, urlparse.netloc will be empty
|
||||||
|
raise InvalidHTTPSpecError(f"Invalid URL: {url}. Ensure it starts with 'http://' or 'https://'")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
# Initialize headers and body
|
# Initialize headers and body
|
||||||
headers = {}
|
headers = {}
|
||||||
body = ""
|
body = ""
|
||||||
|
|||||||
Reference in new issue
Block a user