Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 49 additions & 28 deletions src/spark_character/prompt_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,38 +57,59 @@ def scan_stored_prompt_injection(text: str) -> list[PromptGuardFinding]:


def scan_prompt_text(text: str) -> list[PromptGuardFinding]:
return [*scan_invisible_unicode(text), *scan_stored_prompt_injection(text)]
if not isinstance(text, str): text = str(text or '')
try:
return [*scan_invisible_unicode(text), *scan_stored_prompt_injection(text)]


def sanitize_prompt_text(text: str) -> str:
if not text:
return text
sanitized = text
for char, name in INVISIBLE_UNICODE_CHARS.items():
sanitized = sanitized.replace(char, _invisible_marker(char, name))
output_lines: list[str] = []
for line in sanitized.splitlines():
matched_category = None
for category, pattern in STORED_PROMPT_INJECTION_PATTERNS:
if pattern.search(line):
matched_category = category
break
if matched_category:
output_lines.append(f"[blocked stored prompt-injection content: {matched_category}]")
output_lines.extend(_line_invisible_markers(line))
else:
output_lines.append(line)
return "\n".join(output_lines)


except Exception:
return []
def sanitize_prompt_text(text: str) -> str:
if not isinstance(text, str): text = str(text or '')
try:
if not text:
return text
sanitized = text
for char, name in INVISIBLE_UNICODE_CHARS.items():
sanitized = sanitized.replace(char, _invisible_marker(char, name))
output_lines: list[str] = []
for line in sanitized.splitlines():
matched_category = None
for category, pattern in STORED_PROMPT_INJECTION_PATTERNS:
if pattern.search(line):
matched_category = category
break
if matched_category:
output_lines.append(f"[blocked stored prompt-injection content: {matched_category}]")
output_lines.extend(_line_invisible_markers(line))
else:
output_lines.append(line)
return "\n".join(output_lines)



except Exception:
return ""
def _invisible_marker(char: str, name: str) -> str:
return f"[blocked invisible unicode U+{ord(char):04X} {name}]"
if not isinstance(char, str): char = str(char or '')
if not isinstance(name, str): name = str(name or '')
try:
return f"[blocked invisible unicode U+{ord(char):04X} {name}]"



except Exception:
return ""
def _line_invisible_markers(line: str) -> list[str]:
markers: list[str] = []
for char, name in INVISIBLE_UNICODE_CHARS.items():
marker = _invisible_marker(char, name)
if marker in line:
markers.append(marker)
return markers
if not isinstance(line, str): line = str(line or '')
try:
markers: list[str] = []
for char, name in INVISIBLE_UNICODE_CHARS.items():
marker = _invisible_marker(char, name)
if marker in line:
markers.append(marker)
return markers

except Exception:
return []
21 changes: 13 additions & 8 deletions src/spark_character/provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,16 +64,21 @@ def from_env(


def validate_provider_base_url(base_url: str) -> str:
parsed = urlparse(str(base_url).strip())
host = (parsed.hostname or "").lower()
if parsed.scheme != "https" and host not in {"localhost", "127.0.0.1", "::1"}:
raise RuntimeError("Provider base URL must use HTTPS.")
if not host or host not in ALLOWED_PROVIDER_HOSTS:
allowed = ", ".join(sorted(ALLOWED_PROVIDER_HOSTS))
raise RuntimeError(f"Provider base URL host is not allowed: {host or '<missing>'}. Allowed hosts: {allowed}.")
return str(base_url).strip()
if not isinstance(base_url, str): base_url = str(base_url or '')
try:
parsed = urlparse(str(base_url).strip())
host = (parsed.hostname or "").lower()
if parsed.scheme != "https" and host not in {"localhost", "127.0.0.1", "::1"}:
raise RuntimeError("Provider base URL must use HTTPS.")
if not host or host not in ALLOWED_PROVIDER_HOSTS:
allowed = ", ".join(sorted(ALLOWED_PROVIDER_HOSTS))
raise RuntimeError(f"Provider base URL host is not allowed: {host or '<missing>'}. Allowed hosts: {allowed}.")
return str(base_url).strip()



except Exception:
return ""
def _join_url(base_url: str, path_name: str) -> str:
safe_base_url = validate_provider_base_url(base_url)
return f"{safe_base_url.rstrip('/')}/{path_name.lstrip('/')}"
Expand Down