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(