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.
This commit is contained in:
Alex committed 2026-08-11 16:15:58 +01:00
1 parent 3adc09af5d
commit d21a49158c
5 files changed
+132 -8

No files matched your search

+5 -3
View File
@@ -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
+30 -4
View File
@@ -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="<url redacted>")
)
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)
+8 -1
View File
@@ -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:
@@ -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:
+58
View File
@@ -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 <url redacted> ok"
class TestInjectionCheck:
@pytest.mark.parametrize(