feat(security): integrate canary token into analysis pipeline — closes #352
Adds canary token injection and output safety validation to the full analysis pipeline. Full canary leaks block publishing. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Juan Manuel Servera committed
Jun 12, 2026 at 08:26 UTC
d7fbc73af770caffc5e24edb70c1614d6a94bc48
4 files changed
+133
-8
docs/prompt-injection-guardrails.md
+18
-6
@@ -107,21 +107,23 @@ When creating or modifying prompt templates:
107
108
## Scope
109
110
-This document covers the complete Phase 1 and Phase 2 guardrails for issue #352:
110
+This document covers the complete Phase 1, Phase 2, and pipeline integration guardrails for issue #352:
111
112
- **Phase 1** (complete): Sanitization, boundary fencing, closing constraints, and lint enforcement for all prompt placeholders — including previously semi-trusted variables (`{{WISDOM}}`, `{{SKILLS}}`, `{{WISDOM_CONTENT}}`, `{{TOPIC_DESCRIPTION}}`).
113
- **Phase 2** (complete): Canary token leak detection, red-team corpus testing, and tool evaluation (Garak, LLM Guard, Azure Prompt Shields).
114
+- **Pipeline Integration** (complete): Canary tokens automatically injected in all `call_github_models()` callers (`analyze_fallback.py` and `reskill.py`), output validated via `validate_output_safety()` for canary leaks and boundary marker reproduction. Full canary leak blocks publishing; partial/boundary violations emit warnings.
115
116
### 5. Canary Token Leak Detection (`scripts/canary_token.py`)
117
117
-Each prompt invocation can embed a unique canary token (format: `SQSC-CANARY-<16 hex>`). The token is:
118
+Each prompt invocation embeds a unique canary token (format: `SQSC-CANARY-<16 hex>`). The token is:
119
119
-- Injected into system framing with explicit instructions never to reproduce it
120
+- Automatically injected by `call_github_models()` in both `analyze_fallback.py` and `reskill.py` before sending to the LLM
121
- Unique per invocation (secrets + timestamp) to prevent replay
122
- Checked in generated output via exact, case-insensitive, and partial pattern matching
122
-- Any detection logged at CRITICAL level and raises a security alert
123
+- Full/case-variant leaks logged at CRITICAL level and block publishing
124
+- Partial prefix matches logged at WARNING level and emit a GitHub Actions warning
125
124
-Usage:
126
+Usage (standalone):
127
```python
128
from scripts.canary_token import generate_canary, inject_canary, check_output_for_leak
129
@@ -133,7 +135,17 @@ if result.leaked:
135
raise RuntimeError(f"Canary leaked at position {result.match_position}")
136
```
137
136
-### 6. Red-Team Corpus Testing (`tests/test_redteam_corpus.py`)
138
+### 6. Output Safety Validation (`scripts/analyze_fallback.validate_output_safety`)
139
+
140
+Post-generation validation checks for:
141
+
142
+- **Canary token leaks** — specific token from the current invocation
143
+- **Unknown canary patterns** — catches leaks from prior invocations or cross-contamination
144
+- **Boundary marker reproduction** — detects if the model leaked `<untrusted-content>` or `</untrusted-content>` tags from prompt framing
145
+
146
+This is automatically called after `call_github_models()` returns. Violations emit `::warning::` annotations in CI.
147
+
148
+### 7. Red-Team Corpus Testing (`tests/test_redteam_corpus.py`)
149
150
Automated test suite with 30+ known prompt injection strings across 6 attack categories:
151
scripts/analyze_fallback.py
+59
-1
@@ -1032,11 +1032,58 @@ def validate_https_url(url: str, *, label: str, allowed_hosts: frozenset[str] |
1032
raise ValueError(f"{label} host must be one of {sorted(allowed_hosts)}: {url}")
1033
1034
1035
+def validate_output_safety(output: str, canary: str | None = None) -> list[str]:
1036
+ """Check generated analysis output for canary leaks and injection artifacts.
1037
+
1038
+ Returns a list of security violation messages (empty = safe).
1039
+ """
1040
+ from scripts.canary_token import check_output_for_leak, check_output_for_any_canary
1041
+
1042
+ violations: list[str] = []
1043
+
1044
+ # Check for specific canary leak
1045
+ if canary:
1046
+ result = check_output_for_leak(output, canary)
1047
+ if result.leaked:
1048
+ violations.append(
1049
+ f"Canary token leaked at position {result.match_position}: "
1050
+ f"model may have been manipulated by injected instructions"
1051
+ )
1052
+
1053
+ # Check for any canary pattern (catches leaks from prior invocations)
1054
+ any_result = check_output_for_any_canary(output)
1055
+ if any_result.leaked and (not canary or any_result.canary.lower() != canary.lower()):
1056
+ violations.append(
1057
+ f"Unknown canary pattern '{any_result.canary}' found at position "
1058
+ f"{any_result.match_position}: possible cross-invocation leak"
1059
+ )
1060
+
1061
+ # Check for boundary marker leaks (model reproduced internal framing)
1062
+ from scripts.sanitize_repo_content import BOUNDARY_OPEN, BOUNDARY_CLOSE
1063
+ if BOUNDARY_OPEN in output:
1064
+ violations.append(
1065
+ "Output contains <untrusted-content> boundary marker — "
1066
+ "model may have leaked prompt structure"
1067
+ )
1068
+ if BOUNDARY_CLOSE in output:
1069
+ violations.append(
1070
+ "Output contains </untrusted-content> boundary marker — "
1071
+ "model may have leaked prompt structure"
1072
+ )
1073
+
1074
+ return violations
1075
+
1076
+
1077
def call_github_models(prompt: str) -> str:
1078
token = os.environ.get("GITHUB_TOKEN")
1079
if not token:
1080
raise RuntimeError("GITHUB_TOKEN is required for GitHub Models fallback.")
1081
1082
+ # Inject canary token for output leak detection
1083
+ from scripts.canary_token import generate_canary, inject_canary
1084
+ canary = generate_canary()
1085
+ prompt = inject_canary(prompt, canary)
1086
+
1087
endpoint = os.environ.get("GITHUB_MODELS_ENDPOINT", DEFAULT_MODELS_ENDPOINT)
1088
validate_https_url(endpoint, label="GitHub Models endpoint", allowed_hosts=ALLOWED_MODELS_HOSTS)
1089
model = os.environ.get("GITHUB_MODELS_MODEL", DEFAULT_MODELS_MODEL)
@@ -1063,7 +1110,18 @@ def call_github_models(prompt: str) -> str:
1110
try:
1111
with request.urlopen(req, timeout=timeout) as response: # nosec B310
1112
response_payload = json.load(response)
1066
- return extract_markdown(response_payload)
1113
+ markdown = extract_markdown(response_payload)
1114
+ # Validate output for canary leak and injection artifacts
1115
+ violations = validate_output_safety(markdown, canary)
1116
+ if violations:
1117
+ msg = f"Output safety violations detected: {'; '.join(violations)}"
1118
+ # Full canary leak = prompt injection confirmed; block publishing
1119
+ canary_leaked = any("Canary token leaked" in v for v in violations)
1120
+ if canary_leaked:
1121
+ raise RuntimeError(f"BLOCKED: {msg}")
1122
+ # Partial/boundary leaks are warnings — log but allow
1123
+ print(f"::warning::{msg}", file=sys.stderr)
1124
+ return markdown
1125
except error.HTTPError as exc:
1126
if exc.code not in RETRYABLE_STATUS_CODES or attempt == MAX_RETRIES:
1127
detail = exc.read().decode("utf-8", errors="replace")
scripts/reskill.py
+16
-1
@@ -259,6 +259,12 @@ def call_github_models(prompt: str) -> str:
259
if not token:
260
raise RuntimeError("GITHUB_TOKEN is required for GitHub Models fallback.")
261
262
+ # Inject canary token for output leak detection
263
+ from scripts.canary_token import generate_canary, inject_canary
264
+ from scripts.analyze_fallback import validate_output_safety
265
+ canary = generate_canary()
266
+ prompt = inject_canary(prompt, canary)
267
+
268
endpoint = os.environ.get("GITHUB_MODELS_ENDPOINT", DEFAULT_MODELS_ENDPOINT)
269
validate_https_url(endpoint, label="GitHub Models endpoint", allowed_hosts=ALLOWED_MODELS_HOSTS)
270
model = os.environ.get("GITHUB_MODELS_MODEL", DEFAULT_MODELS_MODEL)
@@ -289,7 +295,16 @@ def call_github_models(prompt: str) -> str:
295
except error.URLError as exc: # pragma: no cover - network failures are environment-specific
296
raise RuntimeError(f"GitHub Models API request failed: {exc.reason}") from exc
297
292
- return extract_markdown(response_payload)
298
+ markdown = extract_markdown(response_payload)
299
+ # Validate output for canary leak and injection artifacts
300
+ violations = validate_output_safety(markdown, canary)
301
+ if violations:
302
+ msg = f"Output safety violations detected: {'; '.join(violations)}"
303
+ canary_leaked = any("Canary token leaked" in v for v in violations)
304
+ if canary_leaked:
305
+ raise RuntimeError(f"BLOCKED: {msg}")
306
+ print(f"::warning::{msg}", file=sys.stderr)
307
+ return markdown
308
309
310
def main(argv: list[str] | None = None) -> int:
tests/test_canary_token.py
+40
@@ -4,6 +4,7 @@ from __future__ import annotations
4
5
import logging
6
7
+from scripts.analyze_fallback import validate_output_safety
8
from scripts.canary_token import (
9
CANARY_PREFIX,
10
check_output_for_any_canary,
@@ -93,3 +94,42 @@ def test_check_empty_output() -> None:
94
canary = generate_canary()
95
result = check_output_for_leak("", canary)
96
assert not result.leaked
97
+
98
+
99
+# ═══════════════════════════════════════════════════════════════════════════════
100
+# validate_output_safety integration tests
101
+# ═══════════════════════════════════════════════════════════════════════════════
102
+
103
+
104
+def test_validate_output_safety_clean() -> None:
105
+ output = "## This Week's Trends\n\nRust and Go dominate this week.\n"
106
+ violations = validate_output_safety(output)
107
+ assert violations == []
108
+
109
+
110
+def test_validate_output_safety_canary_leak() -> None:
111
+ canary = generate_canary()
112
+ output = f"## Analysis\n\nThe internal token is {canary}.\n"
113
+ violations = validate_output_safety(output, canary)
114
+ assert len(violations) >= 1
115
+ assert "Canary token leaked" in violations[0]
116
+
117
+
118
+def test_validate_output_safety_boundary_marker_leak() -> None:
119
+ output = "## Analysis\n\n<untrusted-content>some data</untrusted-content>\n"
120
+ violations = validate_output_safety(output)
121
+ assert len(violations) >= 1
122
+ assert "boundary marker" in violations[0].lower() or "boundary marker" in violations[1].lower()
123
+
124
+
125
+def test_validate_output_safety_unknown_canary_pattern() -> None:
126
+ output = f"Found: {CANARY_PREFIX}-deadbeef12345678 in output\n"
127
+ violations = validate_output_safety(output)
128
+ assert len(violations) >= 1
129
+ assert "Unknown canary pattern" in violations[0]
130
+
131
+
132
+def test_validate_output_safety_no_false_positives_on_normal_hex() -> None:
133
+ output = "Commit hash: abcdef1234567890abcdef\nSHA: deadbeef12345678\n"
134
+ violations = validate_output_safety(output)
135
+ assert violations == []