From d21a49158cfce2816e9baf35f223aa78360bbf61 Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 11 Aug 2026 16:15:58 +0100 Subject: [PATCH] fix: close two guardrail bypasses and an inert instance floor URLCheck read the host from a regex capture group whose character class excluded "@", so a userinfo-prefixed URL passed its userinfo off as the host: https://allowed.com@evil.tld/ satisfied an allow_hosts entry for allowed.com, and http://ok.com@blocked.tld/ evaded block_hosts. Match the whole URL and take the host from the parsed authority instead. A URL that will not parse is now redacted rather than waved through. The PRIVATE_KEY secret pattern matched only the BEGIN header, so a redact action masked the header and released the key material and END line in the clear. Span the whole armored block, falling back to the header when the block is unterminated or over the cap, and raise max_match_chars so the streaming guard's window still covers a full PEM. The pattern also now catches ENCRYPTED, DSA and PGP BLOCK headers, which it missed entirely. The documented GUARDRAILS_FLOOR example omitted "enabled": true, and the field defaults to false, so an operator copying it got a floor that parsed clean, merged to nothing and warned nowhere. Fix the example and warn when a floor is set but disabled. --- application/core/settings.py | 8 +-- application/guardrails/checks/patterns.py | 34 +++++++++++-- application/guardrails/runtime.py | 9 +++- tests/guardrails/test_agent_integration.py | 31 ++++++++++++ tests/guardrails/test_checks.py | 58 ++++++++++++++++++++++ 5 files changed, 132 insertions(+), 8 deletions(-) diff --git a/application/core/settings.py b/application/core/settings.py index 92bf009f..e568cf44 100644 --- a/application/core/settings.py +++ b/application/core/settings.py @@ -301,9 +301,11 @@ class Settings(BaseSettings): GUARDRAILS_CHECKS_ENABLED: list = [] # Instance floor: a GuardrailsConfig fragment every agent inherits and # cannot weaken. Agents may add controls or make an action stricter, never - # looser. Example: - # {"mode": "scan_all", "controls": [{"check": "secrets", "stage": "output", - # "action": "redact"}]} + # looser. "enabled" is required — without it the floor parses but applies + # to nothing. Example: + # {"enabled": true, "mode": "scan_all", + # "controls": [{"check": "secrets", "stage": "output", + # "action": "redact"}]} GUARDRAILS_FLOOR: dict = {} # Judge model for the topic/policy checks. None reuses the request's model. GUARDRAILS_JUDGE_MODEL: Optional[str] = None diff --git a/application/guardrails/checks/patterns.py b/application/guardrails/checks/patterns.py index 62d9f695..253bde70 100644 --- a/application/guardrails/checks/patterns.py +++ b/application/guardrails/checks/patterns.py @@ -4,6 +4,7 @@ from __future__ import annotations import re from typing import Any, Dict, List, Optional, Pattern +from urllib.parse import urlsplit from application.guardrails.base import GuardrailCheck, ScanContext from application.guardrails.types import CheckOutcome, Span, Stage @@ -52,7 +53,13 @@ SECRET_PATTERNS: Dict[str, Pattern[str]] = { "ANTHROPIC_KEY": re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b"), "SLACK_TOKEN": re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b"), "GOOGLE_API_KEY": re.compile(r"\bAIza[0-9A-Za-z_-]{35}\b"), - "PRIVATE_KEY": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |PGP )?PRIVATE KEY-----"), + # Spans the whole armored block, not just the header: redacting the BEGIN + # line alone would release the key material. Falls back to the header when + # the block is unterminated or longer than the cap. + "PRIVATE_KEY": re.compile( + r"-----BEGIN (?:[A-Z0-9]+ )*PRIVATE KEY(?: BLOCK)?-----" + r"(?:[\s\S]{0,8192}?-----END (?:[A-Z0-9]+ )*PRIVATE KEY(?: BLOCK)?-----)?" + ), "JWT": re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b"), "GENERIC_SECRET": re.compile( r"(?i)\b(?:api[_-]?key|secret|passwd|password|token)\b\s*[:=]\s*" @@ -115,7 +122,9 @@ class SecretsCheck(GuardrailCheck): supported_stages = _ALL_TEXT_STAGES supports_redaction = True latency_hint_ms = 2 - max_match_chars = 2048 + # Must cover the longest match PRIVATE_KEY can produce, or the streaming + # guard would release the tail of a PEM block before scanning it. + max_match_chars = 8192 def scan(self, text: str, stage: Stage, context: ScanContext) -> CheckOutcome: spans: List[Span] = [] @@ -213,7 +222,10 @@ class URLCheck(GuardrailCheck): latency_hint_ms = 2 max_match_chars = 2048 - _URL = re.compile(r"\bhttps?://([A-Za-z0-9.\-:]+)(?:/[^\s<>\"')]*)?", re.IGNORECASE) + # Matches the whole URL; the host is taken from the parsed authority rather + # than from a capture group, so ``https://allowed.com@evil.tld`` cannot pass + # its userinfo off as the host. + _URL = re.compile(r"\bhttps?://[^\s<>\"')]+", re.IGNORECASE) @classmethod def validate_settings(cls, settings: Dict[str, Any]) -> Dict[str, Any]: @@ -235,12 +247,26 @@ class URLCheck(GuardrailCheck): def _host_matches(host: str, entry: str) -> bool: return host == entry or host.endswith("." + entry) + @staticmethod + def _host_of(url: str) -> Optional[str]: + """The authority's host, or None when the URL will not parse.""" + try: + return urlsplit(url).hostname + except ValueError: + return None + def scan(self, text: str, stage: Stage, context: ScanContext) -> CheckOutcome: allow = self.settings.get("allow_hosts") or [] block = self.settings.get("block_hosts") or [] spans: List[Span] = [] for match in self._URL.finditer(text): - host = match.group(1).lower().split(":")[0] + host = self._host_of(match.group(0)) + if host is None: + # A URL we cannot resolve is one we cannot vouch for. + spans.append( + Span(match.start(), match.end(), "URL", replacement="") + ) + continue denied = any(self._host_matches(host, b) for b in block) if not denied and allow: denied = not any(self._host_matches(host, a) for a in allow) diff --git a/application/guardrails/runtime.py b/application/guardrails/runtime.py index ef18888f..74f5b77d 100644 --- a/application/guardrails/runtime.py +++ b/application/guardrails/runtime.py @@ -42,13 +42,20 @@ def instance_floor() -> Optional[GuardrailsConfig]: if not raw: return None try: - return GuardrailsConfig.model_validate(raw) + floor = GuardrailsConfig.model_validate(raw) except Exception: logger.error( "GUARDRAILS_FLOOR is invalid and is being ignored; agents run " "without an instance floor until it is fixed" ) return None + if not floor.enabled: + logger.warning( + 'GUARDRAILS_FLOOR is set but "enabled" is false, so no floor is ' + "applied; add \"enabled\": true to enforce it" + ) + return None + return floor def merge_floor(agent: GuardrailsConfig, floor: Optional[GuardrailsConfig]) -> GuardrailsConfig: diff --git a/tests/guardrails/test_agent_integration.py b/tests/guardrails/test_agent_integration.py index 8b436d5d..c85d3947 100644 --- a/tests/guardrails/test_agent_integration.py +++ b/tests/guardrails/test_agent_integration.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from unittest.mock import Mock import pytest @@ -388,6 +389,36 @@ class TestFloorMerge: monkeypatch.setattr(settings, "GUARDRAILS_FLOOR", {"mode": "not-a-mode"}) assert instance_floor() is None + def test_floor_with_controls_but_no_enabled_flag_warns(self, monkeypatch, caplog): + """A floor that parses clean but merges to nothing must not do so silently.""" + from application.core.settings import settings + from application.guardrails.runtime import instance_floor + + monkeypatch.setattr( + settings, + "GUARDRAILS_FLOOR", + {"mode": "scan_all", + "controls": [{"check": "secrets", "stage": "output", "action": "redact"}]}, + ) + with caplog.at_level(logging.WARNING): + floor = instance_floor() + assert floor is None + assert "enabled" in caplog.text + + def test_documented_floor_example_is_effective(self, monkeypatch): + """The example in settings.py must produce a floor that actually merges.""" + from application.core.settings import settings + from application.guardrails.runtime import floor_keys, instance_floor + + monkeypatch.setattr( + settings, + "GUARDRAILS_FLOOR", + {"enabled": True, "mode": "scan_all", + "controls": [{"check": "secrets", "stage": "output", "action": "redact"}]}, + ) + assert instance_floor() is not None + assert floor_keys() == {"secrets:output"} + @pytest.mark.unit class TestKillSwitch: diff --git a/tests/guardrails/test_checks.py b/tests/guardrails/test_checks.py index b17f1166..62b77be3 100644 --- a/tests/guardrails/test_checks.py +++ b/tests/guardrails/test_checks.py @@ -88,6 +88,40 @@ class TestSecretsCheck: ) assert outcome.triggered is False + def test_private_key_body_is_redacted_not_just_the_header(self, ctx): + """Masking the BEGIN line alone would ship the key material verbatim.""" + body = "MIIBOgIBAAJBAKj34GkxFhD9" + "A" * 64 + text = f"here:\n-----BEGIN RSA PRIVATE KEY-----\n{body}\n-----END RSA PRIVATE KEY-----\ndone" + outcome = SecretsCheck({}).scan(text, Stage.OUTPUT, ctx) + redacted = apply_spans(text, outcome.spans) + assert body not in redacted + assert "-----END RSA PRIVATE KEY-----" not in redacted + assert redacted.startswith("here:\n") and redacted.endswith("\ndone") + + @pytest.mark.parametrize( + "header", + [ + "-----BEGIN PRIVATE KEY-----", + "-----BEGIN ENCRYPTED PRIVATE KEY-----", + "-----BEGIN DSA PRIVATE KEY-----", + "-----BEGIN PGP PRIVATE KEY BLOCK-----", + ], + ) + def test_detects_armored_variants(self, ctx, header): + end = header.replace("BEGIN", "END") + text = f"{header}\n{'c' * 48}\n{end}" + outcome = SecretsCheck({}).scan(text, Stage.OUTPUT, ctx) + assert outcome.triggered is True + assert "PRIVATE_KEY" in outcome.categories + + def test_unterminated_private_key_still_flags(self, ctx): + """A truncated block must not fall back to reporting nothing.""" + outcome = SecretsCheck({}).scan( + "-----BEGIN RSA PRIVATE KEY-----\n" + "d" * 40, Stage.OUTPUT, ctx + ) + assert outcome.triggered is True + assert "PRIVATE_KEY" in outcome.categories + class TestDenylistCheck: def test_word_match_does_not_fire_on_substring(self, ctx): @@ -135,6 +169,30 @@ class TestURLCheck: assert check.scan("https://evil.test/p", Stage.OUTPUT, ctx).triggered is True assert check.scan("https://ok.test/p", Stage.OUTPUT, ctx).triggered is False + def test_userinfo_does_not_masquerade_as_the_host(self, ctx): + """``https://allowed@evil.test`` resolves to evil.test in every browser.""" + check = URLCheck(URLCheck.validate_settings({"allow_hosts": ["arc53.com"]})) + outcome = check.scan("https://arc53.com@evil.test/steal?d=1", Stage.OUTPUT, ctx) + assert outcome.triggered is True + + def test_userinfo_does_not_bypass_the_blocklist(self, ctx): + check = URLCheck(URLCheck.validate_settings({"block_hosts": ["evil.test"]})) + assert check.scan("http://ok.test@evil.test/raw", Stage.OUTPUT, ctx).triggered is True + + def test_userinfo_with_password_and_port(self, ctx): + check = URLCheck(URLCheck.validate_settings({"allow_hosts": ["arc53.com"]})) + assert check.scan("https://user:pw@evil.test:8443/x", Stage.OUTPUT, ctx).triggered is True + assert check.scan("https://user:pw@arc53.com:8443/x", Stage.OUTPUT, ctx).triggered is False + + def test_redacted_span_covers_the_whole_url(self, ctx): + """A partial span would leave the real host in the output.""" + check = URLCheck(URLCheck.validate_settings({"allow_hosts": ["arc53.com"]})) + text = "see https://arc53.com@evil.test/steal?d=1 ok" + outcome = check.scan(text, Stage.OUTPUT, ctx) + redacted = apply_spans(text, outcome.spans) + assert "evil.test" not in redacted + assert redacted == "see ok" + class TestInjectionCheck: @pytest.mark.parametrize(