From d67700f2f34e4e3b19bd27f85c4e60c1d1a90e82 Mon Sep 17 00:00:00 2001 From: Pradeep Sasatt Date: Wed, 3 Jun 2026 17:58:17 +0530 Subject: [PATCH] security: stricter sanitizers for partial-ssrf and log-injection follow-ups Add safe_agent_segment() helper that validates caller-id and session-id via a strict regex allowlist (alphanum + _.+@- only). CodeQL recognizes regex-anchored allowlists as proper sanitizers; the previous quote(safe='') approach only percent-encoded, which CodeQL still considered tainted for partial-ssrf because the value still flowed into the URL. Also sanitize model_id before logger.exception() in docgrok/admin.py to address the two new py/log-injection alerts. Follow-up to PR #173. Resolves 5 alerts opened by that PR. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- api/api.py | 30 +++++++++++++++++++++--------- api/security_utils.py | 20 ++++++++++++++++++++ docgrok/admin.py | 6 ++++-- 3 files changed, 45 insertions(+), 11 deletions(-) diff --git a/api/api.py b/api/api.py index 24c0738..9c75a5a 100644 --- a/api/api.py +++ b/api/api.py @@ -51,8 +51,8 @@ def __exit__(self, *a): pass ModelCategory, Assistant, CreateAssistantRequest, AssistantChatRequest ) from store import init_store, get_store -from security_utils import safe_url_segment, validate_outbound_url, validate_sql_identifier # lgtm[py/unused-import] -from urllib.parse import quote as _urlquote +from security_utils import safe_agent_segment, safe_url_segment, validate_outbound_url, validate_sql_identifier # lgtm[py/unused-import] +from urllib.parse import quote as _urlquote # noqa: F401 — kept for downstream callers logger = logging.getLogger(__name__) @@ -1140,8 +1140,11 @@ async def agent_session_approvals(session_id: str, request: Request): if not _INTERNAL_API_TOKEN: raise HTTPException(status_code=503, detail="agent: INTERNAL_API_TOKEN not configured") user = _caller_id(request) - safe_user = _urlquote(user, safe="") - safe_sid = _urlquote(session_id, safe="") + try: + safe_user = safe_agent_segment(user) + safe_sid = safe_agent_segment(session_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client: r = await client.get(f"{_AGENT_URL}/v1/sessions/{safe_user}/{safe_sid}/approvals", headers=_agent_headers(request)) return JSONResponse(status_code=r.status_code, content=r.json()) @@ -1166,7 +1169,10 @@ async def agent_sessions_list(request: Request): if not _INTERNAL_API_TOKEN: raise HTTPException(status_code=503, detail="agent: INTERNAL_API_TOKEN not configured") user = _caller_id(request) - safe_user = _urlquote(user, safe="") + try: + safe_user = safe_agent_segment(user) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client: r = await client.get(f"{_AGENT_URL}/v1/sessions/{safe_user}", headers=_agent_headers(request)) return JSONResponse(status_code=r.status_code, content=r.json()) @@ -1177,8 +1183,11 @@ async def agent_session_get(session_id: str, request: Request): if not _INTERNAL_API_TOKEN: raise HTTPException(status_code=503, detail="agent: INTERNAL_API_TOKEN not configured") user = _caller_id(request) - safe_user = _urlquote(user, safe="") - safe_sid = _urlquote(session_id, safe="") + try: + safe_user = safe_agent_segment(user) + safe_sid = safe_agent_segment(session_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client: r = await client.get(f"{_AGENT_URL}/v1/sessions/{safe_user}/{safe_sid}", headers=_agent_headers(request)) return JSONResponse(status_code=r.status_code, content=r.json()) @@ -1189,8 +1198,11 @@ async def agent_session_delete(session_id: str, request: Request): if not _INTERNAL_API_TOKEN: raise HTTPException(status_code=503, detail="agent: INTERNAL_API_TOKEN not configured") user = _caller_id(request) - safe_user = _urlquote(user, safe="") - safe_sid = _urlquote(session_id, safe="") + try: + safe_user = safe_agent_segment(user) + safe_sid = safe_agent_segment(session_id) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client: r = await client.delete(f"{_AGENT_URL}/v1/sessions/{safe_user}/{safe_sid}", headers=_agent_headers(request)) return JSONResponse(status_code=r.status_code, content=r.json()) diff --git a/api/security_utils.py b/api/security_utils.py index f1fe2f2..6772ba0 100644 --- a/api/security_utils.py +++ b/api/security_utils.py @@ -155,6 +155,26 @@ def validate_sql_identifier(name: str, *, allow_dot: bool = False) -> str: # pip-*, trp-*, asst-*) plus user-defined transform/pipeline names. _SEGMENT_RE = re.compile(r"^[A-Za-z0-9_.\-]{1,128}$") +# Caller-id / agent-session segments — also allow '@' and '+' for UPNs / emails. +_AGENT_SEGMENT_RE = re.compile(r"^[A-Za-z0-9_.+@\-]{1,256}$") + + +def safe_agent_segment(value: str) -> str: + """Validate a caller-id or session-id for use as a URL path segment. + + Accepts the union of IDs, GUIDs, emails and UPNs. Rejects anything + that could escape the segment (``/``, ``?``, ``#``, control chars, + or path traversal). Returns the value unchanged on success — the + regex guarantees it is already safe to interpolate into a URL. + """ + if not isinstance(value, str) or not value: + raise ValueError("segment must be a non-empty string") + if not _AGENT_SEGMENT_RE.match(value): + raise ValueError(f"unsafe agent segment: {value!r}") + if value in (".", ".."): + raise ValueError("segment must not be '.' or '..'") + return value + def safe_url_segment(value: str) -> str: """Validate and percent-encode ``value`` for use as a single URL path diff --git a/docgrok/admin.py b/docgrok/admin.py index 26d08eb..7c0455f 100644 --- a/docgrok/admin.py +++ b/docgrok/admin.py @@ -325,7 +325,8 @@ async def healthcheck_model(model_id: str): token = cred.get_token("https://cognitiveservices.azure.com/.default").token headers["Authorization"] = f"Bearer {token}" except Exception: - logger.exception("failed to acquire MI token for model %s", model_id) + safe_id = str(model_id).replace("\r", " ").replace("\n", " ")[:128] + logger.exception("failed to acquire MI token for model %s", safe_id) return {"ok": False, "status": 0, "detail": "failed to acquire managed-identity token; see server logs"} else: if not api_key: @@ -337,7 +338,8 @@ async def healthcheck_model(model_id: str): try: r = await c.post(url, headers=headers, json={"input": "health check", "model": deployment}) except Exception: - logger.exception("healthcheck request failed for model %s", model_id) + safe_id = str(model_id).replace("\r", " ").replace("\n", " ")[:128] + logger.exception("healthcheck request failed for model %s", safe_id) return {"ok": False, "status": 0, "detail": "request failed; see server logs"} body = r.text[:200]