dev
278ecca018
Keeps credentials, company names, personal names and PII out of the model's context. Enforcement lives in Claude Code hooks rather than in instructions to the agent: a skill alone cannot protect anything, because by the time the agent reads a rule the surrounding context has already been sent. Credentials are removed irreversibly and marked. Entities from a user-supplied dictionary become stable aliases, rewritten back to real values on their way to disk and to the shell, so code and commands referring to them still work. Published from a clean tree; development history is not included.
540 lines
23 KiB
Python
540 lines
23 KiB
Python
"""Engine unit tests. Run: python3 tests/unit/test_engine.py"""
|
||
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts"))
|
||
|
||
_TMP = tempfile.mkdtemp(prefix="ctxguard-test-")
|
||
os.environ["CTXGUARD_HOME"] = _TMP
|
||
|
||
from ctxguard import detect, policy as policy_mod, substitute
|
||
from ctxguard.store import Store, base26
|
||
|
||
SLUG = "test-project"
|
||
|
||
|
||
def engine(entities=()):
|
||
store = Store(SLUG)
|
||
pol = policy_mod.Policy()
|
||
for value, type_, match, hint in entities:
|
||
store.add(value, type_, match=match, hint=hint, prefix=pol.alias_prefix)
|
||
return substitute.Engine(store, pol), store
|
||
|
||
|
||
class TestSecrets(unittest.TestCase):
|
||
def test_vendor_keys_are_replaced_by_markers(self):
|
||
eng, _ = engine()
|
||
cases = [
|
||
"AKIA1234567890ABCDEF",
|
||
"ghp_Rf2Lp9Zt4Wm1Nb8Kd3Vy6Qs5Hj7Xc0Ae1Bg9",
|
||
"sk-ant-api03-Rf2Lp9Zt4Wm1Nb8Kd3Vy6Qs5Hj7Xc0",
|
||
"glpat-Rf2Lp9Zt4Wm1Nb8Kd3Vy",
|
||
"xoxb-123456789012-Rf2Lp9Zt4Wm1",
|
||
]
|
||
for secret in cases:
|
||
with self.subTest(secret=secret[:8]):
|
||
out = eng.redact(f"key = {secret} end")
|
||
self.assertNotIn(secret, out.text)
|
||
self.assertIn("<SECRET:", out.text)
|
||
|
||
def test_marker_is_stable_and_carries_no_value(self):
|
||
eng, _ = engine()
|
||
text = "a = AKIA1234567890ABCDEF and b = AKIA1234567890ABCDEF"
|
||
out = eng.redact(text)
|
||
markers = substitute.MARKER_RE.findall(out.text)
|
||
self.assertEqual(len(markers), 2)
|
||
self.assertEqual(markers[0], markers[1], "same secret must map to the same marker")
|
||
|
||
def test_private_key_block(self):
|
||
eng, _ = engine()
|
||
out = eng.redact("-----BEGIN RSA PRIVATE KEY-----\nMIIE...\n")
|
||
self.assertIn("<SECRET:private-key-block:", out.text)
|
||
|
||
def test_connection_string_password(self):
|
||
eng, _ = engine()
|
||
out = eng.redact("DSN=postgres://admin:sup3rS3cret@db.internal:5432/app")
|
||
self.assertNotIn("sup3rS3cret", out.text)
|
||
|
||
def test_local_dev_connection_strings_are_not_secrets(self):
|
||
"""Regression from real data: 265 of 265 credential-URL hits in one session
|
||
were user==password on loopback. Firing there gets the plugin disabled."""
|
||
eng, _ = engine()
|
||
for dsn in (
|
||
"postgres://service:service@127.0.0.1:5432/app",
|
||
"postgres://app:app@db:5432/app",
|
||
"mysql://root:root@localhost/test",
|
||
"redis://user:password@127.0.0.1:6379",
|
||
"postgres://svc:postgres@db/app",
|
||
):
|
||
with self.subTest(dsn=dsn):
|
||
self.assertEqual(eng.redact(dsn).secret_count, 0)
|
||
|
||
def test_real_connection_string_still_caught_on_loopback(self):
|
||
eng, _ = engine()
|
||
out = eng.redact("postgres://svc:Xq7Rf2Lp9Zt4Wm@127.0.0.1:5432/app")
|
||
self.assertEqual(out.secret_count, 1)
|
||
|
||
def test_documented_examples_are_not_blocked(self):
|
||
eng, _ = engine()
|
||
out = eng.redact("use AKIAIOSFODNN7EXAMPLE in the docs")
|
||
self.assertEqual(out.secret_count, 0)
|
||
|
||
def test_allowlist_entry_inside_a_live_token_does_not_launder_it(self):
|
||
"""Regression: a bidirectional allowlist check let 'example.com' inside a
|
||
real token suppress the finding."""
|
||
eng, _ = engine()
|
||
secret = "AKIAEXAMPLECOM12345Z"
|
||
out = eng.redact(f"key = {secret}")
|
||
self.assertNotIn(secret, out.text)
|
||
|
||
def test_low_entropy_assignment_is_not_a_secret(self):
|
||
eng, _ = engine()
|
||
out = eng.redact('password = "aaaaaaaaaaaaaaaaaaaa"')
|
||
self.assertEqual(out.secret_count, 0)
|
||
|
||
def test_prefixed_env_var_secrets_are_caught(self):
|
||
"""Regression from a real .env: `_` is a word character, so `\\bapi_key\\b`
|
||
never fired inside SERVICE_API_KEY and four of five common shapes were missed."""
|
||
eng, _ = engine()
|
||
for line in (
|
||
"SERVICE_API_KEY=Xq7Rf2Lp9Zt4W",
|
||
"SECRET_KEY=Xq7Rf2Lp9Zt4W",
|
||
"APP_DATABASE_PASSWORD=Xq7Rf2Lp9Zt4W",
|
||
"MY_AUTH_TOKEN=Xq7Rf2Lp9Zt4W",
|
||
"api_key=Xq7Rf2Lp9Zt4W",
|
||
):
|
||
with self.subTest(line=line):
|
||
self.assertEqual(eng.redact(line).secret_count, 1)
|
||
|
||
def test_short_secret_is_caught_despite_entropy_ceiling(self):
|
||
"""Entropy is bounded by log2(len), so a 13-char key can never reach 3.5.
|
||
Character-class diversity has to carry short values."""
|
||
eng, _ = engine()
|
||
self.assertEqual(eng.redact("SERVICE_API_KEY=ab3-xy9-kd2m1").secret_count, 1)
|
||
|
||
def test_empty_config_key_does_not_capture_the_next_line(self):
|
||
"""Regression from a real .env: `\\s*` after the separator crossed newlines, so
|
||
an empty key swallowed the following line as its value."""
|
||
eng, _ = engine()
|
||
text = "CONTOSO_API_KEY=\nCONTOSO_SECRET_KEY=\nTIMEOUT_MAX_MS=200\n"
|
||
self.assertEqual(eng.redact(text).secret_count, 0)
|
||
|
||
def test_code_references_are_not_secrets(self):
|
||
"""Regression: `SET password_hash = EXCLUDED.password_hash` in real SQL."""
|
||
eng, _ = engine()
|
||
for line in (
|
||
"SET password_hash = EXCLUDED.password_hash,",
|
||
"let token = config.auth_token;",
|
||
"api_key: settings.routing.api_key",
|
||
"password = ${DB_PASSWORD}",
|
||
"api_key = <your-key>",
|
||
):
|
||
with self.subTest(line=line):
|
||
self.assertEqual(eng.redact(line).secret_count, 0)
|
||
|
||
def test_dev_password_assignment_is_not_a_secret(self):
|
||
eng, _ = engine()
|
||
self.assertEqual(eng.redact("POSTGRES_PASSWORD: service").secret_count, 0)
|
||
|
||
def test_high_entropy_assignment_is_a_secret(self):
|
||
eng, _ = engine()
|
||
out = eng.redact('api_key = "Xq7Rf2Lp9Zt4Wm1Nb8Kd3Vy6"')
|
||
self.assertEqual(out.secret_count, 1)
|
||
|
||
def test_own_markers_are_not_detected_as_secrets(self):
|
||
"""Regression: the word SECRET inside `<SECRET:rule:hash>` matched the keyword
|
||
rule, so the leak metric counted the engine's own redactions and could never
|
||
reach zero."""
|
||
eng, _ = engine()
|
||
marked = "SERVICE_API_KEY: <SECRET:generic-assignment:cafda980>"
|
||
self.assertEqual(eng.redact(marked).secret_count, 0)
|
||
self.assertEqual(eng.redact(marked).text, marked)
|
||
|
||
def test_rule_names_in_prose_are_not_secrets(self):
|
||
"""A marker quoted with the hash elided -- `<SECRET:generic-assignment:...>` --
|
||
does not match MARKER_RE, so the rule name inside read as a credential."""
|
||
eng, _ = engine()
|
||
for text in ("a handful of `<SECRET:generic-assignment:…>` markers",
|
||
"rules fired: secret:aws-access-key"):
|
||
with self.subTest(text=text):
|
||
self.assertEqual(eng.redact(text).secret_count, 0)
|
||
|
||
def test_our_own_report_format_is_not_self_detected(self):
|
||
"""`secret:generic-assignment=85` read as keyword-colon-value, so piping a
|
||
report through the shell filter redacted its own labels."""
|
||
from ctxguard.cli import _render
|
||
eng, _ = engine()
|
||
report = _render({"secret:generic-assignment": 85, "secret:aws-access-key": 3})
|
||
self.assertEqual(eng.redact(report).secret_count, 0)
|
||
self.assertEqual(eng.redact(report).text, report)
|
||
|
||
def test_redacting_twice_is_stable(self):
|
||
eng, _ = engine()
|
||
once = eng.redact("k = AKIA2X7QP4LMZ9KDVR3T").text
|
||
twice = eng.redact(once).text
|
||
self.assertEqual(once, twice)
|
||
|
||
def test_secrets_are_irreversible(self):
|
||
eng, _ = engine()
|
||
secret = "AKIA1234567890ABCDEF"
|
||
redacted = eng.redact(f"k={secret}").text
|
||
restored, _ = eng.rehydrate(redacted)
|
||
self.assertNotIn(secret, restored)
|
||
self.assertTrue(eng.has_secret_marker(restored))
|
||
|
||
|
||
class TestEntities(unittest.TestCase):
|
||
def setUp(self):
|
||
self.eng, self.store = engine([
|
||
("Acme Corp", "company", "exact", "retail customer"),
|
||
("Иван Петров", "person", "stem", "backend developer"),
|
||
("db.acme.internal", "host", "exact", "prod database"),
|
||
])
|
||
|
||
def test_company_alias(self):
|
||
out = self.eng.redact("Acme Corp asked for a report")
|
||
self.assertIn("CTXG_COMPANY_A", out.text)
|
||
self.assertNotIn("Acme", out.text)
|
||
|
||
def test_homoglyph_evasion_is_caught(self):
|
||
out = self.eng.redact("Aсme Corp") # Cyrillic 'с'
|
||
self.assertIn("CTXG_COMPANY_A", out.text)
|
||
|
||
def test_separator_variants(self):
|
||
for variant in ("acme-corp", "AcmeCorp", "ACME_CORP"):
|
||
with self.subTest(variant=variant):
|
||
self.assertIn("CTXG_COMPANY_A", self.eng.redact(variant).text)
|
||
|
||
def test_russian_declensions(self):
|
||
for form in ("Иван Петров", "Ивана Петрова", "Ивану Петрову", "Иваном Петровым"):
|
||
with self.subTest(form=form):
|
||
out = self.eng.redact(f"Задачу закрыл {form}.")
|
||
self.assertIn("CTXG_PERSON_A", out.text)
|
||
self.assertNotIn("Петров", out.text)
|
||
|
||
def test_longest_match_wins(self):
|
||
self.store.add("Acme", "company", prefix="CTXG_")
|
||
eng = substitute.Engine(self.store, policy_mod.Policy())
|
||
out = eng.redact("Acme Corp")
|
||
self.assertIn("CTXG_COMPANY_A", out.text)
|
||
self.assertNotIn("CTXG_COMPANY_B", out.text)
|
||
|
||
def test_roundtrip(self):
|
||
original = "Иван Петров из Acme Corp правит db.acme.internal"
|
||
redacted = self.eng.redact(original).text
|
||
restored, n = self.eng.rehydrate(redacted)
|
||
self.assertEqual(restored, original)
|
||
self.assertEqual(n, 3)
|
||
|
||
def test_glossary_never_leaks_values(self):
|
||
blob = repr(self.store.glossary())
|
||
for forbidden in ("Acme", "Петров", "db.acme.internal"):
|
||
self.assertNotIn(forbidden, blob)
|
||
|
||
def test_hint_repeating_the_value_is_rejected(self):
|
||
with self.assertRaises(ValueError):
|
||
self.store.add("Globex", "company", hint="Globex is a customer")
|
||
|
||
|
||
class TestIdentMode(unittest.TestCase):
|
||
"""Regression from a live run: the real vendor name leaked through code
|
||
identifiers, where boundary-anchored patterns cannot match."""
|
||
|
||
def setUp(self):
|
||
self.eng, self.store = engine([("Contoso", "company", "ident", "address vendor")])
|
||
|
||
def test_matches_inside_identifiers(self):
|
||
for text in ("ContosoClient", "struct ContosoConfig {", "CONTOSO_API_KEY=x",
|
||
"src/clients/contoso.rs", "use crate::contoso::Client;"):
|
||
with self.subTest(text=text):
|
||
out = self.eng.redact(text).text
|
||
self.assertNotIn("adata", out, f"real name survived in {out!r}")
|
||
|
||
def test_exact_mode_would_have_missed_these(self):
|
||
store = Store(SLUG)
|
||
store.add("Contoso", "company", match="exact", prefix="CTXG_")
|
||
eng = substitute.Engine(store, policy_mod.Policy())
|
||
self.assertIn("ContosoClient", eng.redact("ContosoClient").text)
|
||
|
||
def test_case_survives_the_round_trip(self):
|
||
for original in ("ContosoClient", "CONTOSO_API_KEY", "contoso.rs",
|
||
"Contoso and CONTOSO and contoso"):
|
||
with self.subTest(original=original):
|
||
red = self.eng.redact(original).text
|
||
restored, n = self.eng.rehydrate(red)
|
||
self.assertEqual(restored, original)
|
||
self.assertGreater(n, 0)
|
||
|
||
def test_cli_exposes_the_mode(self):
|
||
"""The engine supported ident before the CLI did, so it was unreachable."""
|
||
from ctxguard.cli import build_parser
|
||
args = build_parser().parse_args(
|
||
["entity", "add", "Contoso", "--type", "company", "--match", "ident"])
|
||
self.assertEqual(args.match, "ident")
|
||
|
||
def test_short_values_are_refused(self):
|
||
with self.assertRaises(ValueError):
|
||
self.store.add("ACL", "project", match="ident")
|
||
|
||
|
||
class TestPII(unittest.TestCase):
|
||
def test_email_is_reversible(self):
|
||
eng, _ = engine()
|
||
original = "write to ivan.petrov@acme.ru please"
|
||
red = eng.redact(original)
|
||
self.assertNotIn("ivan.petrov@acme.ru", red.text)
|
||
self.assertIn("CTXG_EMAIL_", red.text)
|
||
restored, n = eng.rehydrate(red.text)
|
||
self.assertEqual(restored, original)
|
||
self.assertEqual(n, 1)
|
||
|
||
def test_same_email_same_alias(self):
|
||
eng, _ = engine()
|
||
out = eng.redact("a@b.ru and a@b.ru").text
|
||
aliases = {t for t in out.split() if t.startswith("CTXG_EMAIL_")}
|
||
self.assertEqual(len(aliases), 1)
|
||
|
||
def test_russian_phone(self):
|
||
eng, _ = engine()
|
||
for form in ("+7 916 123-45-67", "+79161234567", "8 (916) 123-45-67"):
|
||
with self.subTest(form=form):
|
||
self.assertIn("CTXG_PHONE_", eng.redact(f"тел {form}").text)
|
||
|
||
def test_card_requires_luhn(self):
|
||
eng, _ = engine()
|
||
self.assertIn("CTXG_ID_", eng.redact("card 4111111111111111").text)
|
||
self.assertNotIn("CTXG_ID_", eng.redact("id 4111111111111112").text)
|
||
|
||
def test_card_requires_a_real_issuer_prefix(self):
|
||
"""Regression: Luhn alone passes 1 in 10 random digit runs, so lockfile
|
||
checksums and base64 blobs were reported as payment cards."""
|
||
eng, _ = engine()
|
||
# Luhn-valid but no issuer ever started with 9 or 1
|
||
self.assertNotIn("CTXG_ID_", eng.redact("n 9999999999999995").text)
|
||
self.assertNotIn("CTXG_ID_", eng.redact("n 1234567890128").text)
|
||
for card in ("4111111111111111", "5500005555555559", "371449635398431"):
|
||
with self.subTest(card=card):
|
||
self.assertIn("CTXG_ID_", eng.redact(f"card {card}").text)
|
||
|
||
def test_digits_inside_a_hex_token_are_not_a_card(self):
|
||
eng, _ = engine()
|
||
self.assertNotIn("CTXG_ID_", eng.redact("sha=ab4111111111111111cd").text)
|
||
|
||
def test_inn_needs_context_keyword(self):
|
||
eng, _ = engine()
|
||
self.assertIn("CTXG_ID_", eng.redact("ИНН 7712345678").text)
|
||
self.assertNotIn("CTXG_ID_", eng.redact("count 7712345678").text)
|
||
|
||
|
||
class TestStructural(unittest.TestCase):
|
||
def setUp(self):
|
||
self.pol = policy_mod.Policy()
|
||
|
||
def test_env_file_denied(self):
|
||
for path in ("/srv/app/.env", ".env", "/srv/app/.env.production", "/x/id_rsa"):
|
||
with self.subTest(path=path):
|
||
self.assertIsNotNone(detect.denied_path(path, self.pol))
|
||
|
||
def test_documented_templates_are_readable(self):
|
||
"""Regression from a live run: the .env refusal told the agent to read
|
||
.env.example, and the next rule blocked that too. Contradictory guidance is
|
||
how an agent learns to ignore guidance."""
|
||
for path in ("/srv/app/.env.example", ".env.sample", "/srv/app/.env.template"):
|
||
with self.subTest(path=path):
|
||
self.assertIsNone(detect.denied_path(path, self.pol))
|
||
|
||
def test_cat_env_example_is_allowed_but_cat_env_is_not(self):
|
||
self.assertIsNone(detect.denied_command("cat .env.example", self.pol))
|
||
self.assertIsNone(detect.denied_command("cat .env.sample", self.pol))
|
||
self.assertIsNotNone(detect.denied_command("cat .env", self.pol))
|
||
self.assertIsNotNone(detect.denied_command("cat config/.env", self.pol))
|
||
|
||
def test_ordinary_source_allowed(self):
|
||
for path in ("/srv/app/main.py", "README.md", "/srv/app/env.example.md"):
|
||
with self.subTest(path=path):
|
||
self.assertIsNone(detect.denied_path(path, self.pol))
|
||
|
||
def test_self_protection(self):
|
||
for path in (
|
||
"/home/u/.claude/settings.json",
|
||
"/home/u/.claude/ctx-guard/projects/x/entities.json",
|
||
"/repo/plugins/ctxguard/scripts/ctxguard/detect.py",
|
||
):
|
||
with self.subTest(path=path):
|
||
self.assertIsNotNone(detect.self_protected(path, self.pol))
|
||
|
||
def test_credential_dumping_commands_denied(self):
|
||
for cmd in ("env", "printenv", "gh auth token", "kubectl get secret app -o yaml",
|
||
"cat .env", "aws configure get aws_secret_access_key", "op read op://a/b",
|
||
"env | grep TOKEN", "printenv SERVICE_API_KEY", "sudo env",
|
||
"ls; env", "cat config/.env"):
|
||
with self.subTest(cmd=cmd):
|
||
self.assertIsNotNone(detect.denied_command(cmd, self.pol))
|
||
|
||
def test_ordinary_commands_allowed(self):
|
||
for cmd in ("ls -la", "git status", "npm test", "grep -r foo src/",
|
||
"cat .env.example", "ls -la .env", "cp .env.example .env",
|
||
"grep ROUTING src/config.rs"):
|
||
with self.subTest(cmd=cmd):
|
||
self.assertIsNone(detect.denied_command(cmd, self.pol))
|
||
|
||
|
||
class TestPolicyPersistence(unittest.TestCase):
|
||
"""Regression: `init` wrote a full snapshot of every rule list, so a later plugin
|
||
upgrade shipped fixed rules the hook never read. The fix appeared not to work and
|
||
only a direct comparison of file vs code showed why."""
|
||
|
||
def test_defaults_persist_as_a_diff(self):
|
||
import json
|
||
stored = json.loads(policy_mod.Policy().to_json())
|
||
self.assertEqual(stored, {"version": policy_mod.POLICY_VERSION})
|
||
|
||
def test_customisation_persists(self):
|
||
import json
|
||
pol = policy_mod.Policy()
|
||
pol.mode = "observe"
|
||
pol.allowlist = pol.allowlist + ["MY-OWN-EXAMPLE"]
|
||
stored = json.loads(pol.to_json())
|
||
self.assertEqual(stored["mode"], "observe")
|
||
self.assertIn("MY-OWN-EXAMPLE", stored["allowlist"])
|
||
self.assertNotIn("deny_command_patterns", stored)
|
||
|
||
def test_stale_v1_rule_lists_are_discarded(self):
|
||
stale = {
|
||
"version": 1,
|
||
"mode": "observe",
|
||
"deny_command_patterns": [r"(?<![\w-])(env|printenv)(?![\w-])(?!.*\|)"],
|
||
"deny_read_paths": ["**/.env", "**/.env.*"],
|
||
}
|
||
pol = policy_mod.Policy.from_dict(stale)
|
||
self.assertEqual(pol.mode, "observe", "preferences must survive")
|
||
self.assertEqual(pol.version, policy_mod.POLICY_VERSION)
|
||
# the fixed rules must apply, not the snapshot
|
||
self.assertIsNone(detect.denied_command("cat .env.example", pol))
|
||
self.assertIsNone(detect.denied_command("ls -la .env", pol))
|
||
self.assertIsNotNone(detect.denied_command("env", pol))
|
||
self.assertIsNone(detect.denied_path("/srv/.env.example", pol))
|
||
|
||
def test_dev_credentials_can_be_made_strict(self):
|
||
pol = policy_mod.Policy()
|
||
pol.dev_credentials_are_secrets = True
|
||
store = Store(SLUG)
|
||
eng = substitute.Engine(store, pol)
|
||
self.assertEqual(
|
||
eng.redact("postgres://app:app@127.0.0.1/db").secret_count, 1)
|
||
|
||
|
||
class TestDictionaryHandover(unittest.TestCase):
|
||
"""Aliases cannot be reproduced: they are assigned in registration order, and the
|
||
salt is per-store. Two people who each ran `entity add` would mean different
|
||
companies by CTXG_COMPANY_A, and their secret markers would not correlate. So the
|
||
dictionary has to be copied, and copying it has to reproduce output exactly."""
|
||
|
||
def test_aliases_are_order_dependent(self):
|
||
a = Store("proj")
|
||
a.add("Globex", "company", prefix="CTXG_")
|
||
a.add("Initech", "company", prefix="CTXG_")
|
||
b = Store("proj")
|
||
b.add("Initech", "company", prefix="CTXG_")
|
||
b.add("Globex", "company", prefix="CTXG_")
|
||
self.assertNotEqual(
|
||
a.find("Globex").alias, b.find("Globex").alias,
|
||
"if this ever passes, the handover warning can be relaxed",
|
||
)
|
||
|
||
def test_salt_differs_per_store(self):
|
||
self.assertNotEqual(Store("p").salt, Store("p").salt)
|
||
|
||
def test_copied_dictionary_reproduces_output_exactly(self):
|
||
a = Store("proj")
|
||
a.add("Globex", "company", match="ident", prefix="CTXG_")
|
||
a.add("Иван Петров", "person", match="stem", prefix="CTXG_")
|
||
|
||
# what export/import moves: salt, counters, entities
|
||
b = Store("proj")
|
||
b.salt = a.salt
|
||
b.counters = dict(a.counters)
|
||
b.entities = list(a.entities)
|
||
b._invalidate()
|
||
|
||
pol = policy_mod.Policy()
|
||
text = ("Globex и GlobexClient, закрыл Иваном Петровым, "
|
||
"key=AKIA2X7QP4LMZ9KDVR3T, mail ivan@globex.ru")
|
||
out_a = substitute.Engine(a, pol).redact(text).text
|
||
out_b = substitute.Engine(b, pol).redact(text).text
|
||
self.assertEqual(out_a, out_b)
|
||
self.assertIn("<SECRET:aws-access-key:", out_a)
|
||
self.assertIn("CTXG_EMAIL_", out_a)
|
||
|
||
def test_export_refuses_to_land_in_a_git_repo(self):
|
||
from ctxguard.cli import _inside_git_worktree
|
||
self.assertTrue(_inside_git_worktree(Path(__file__)))
|
||
|
||
|
||
class TestProjectIdentity(unittest.TestCase):
|
||
"""Regression: keying the dictionary by cwd meant a session started in a
|
||
subdirectory got a different, empty dictionary -- protection that looked enabled
|
||
and applied to nothing."""
|
||
|
||
def test_subdirectories_share_one_dictionary(self):
|
||
import subprocess
|
||
from ctxguard import paths as pmod
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
root = Path(tmp) / "repo"
|
||
(root / "src" / "deep").mkdir(parents=True)
|
||
subprocess.run(["git", "init", "-q", str(root)], check=True,
|
||
capture_output=True)
|
||
pmod._TOPLEVEL_CACHE.clear()
|
||
slugs = {pmod.slug(root), pmod.slug(root / "src"),
|
||
pmod.slug(root / "src" / "deep")}
|
||
self.assertEqual(len(slugs), 1, f"expected one slug, got {slugs}")
|
||
|
||
def test_non_repository_falls_back_to_the_directory(self):
|
||
from ctxguard import paths as pmod
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
a, b = Path(tmp) / "a", Path(tmp) / "b"
|
||
a.mkdir(); b.mkdir()
|
||
pmod._TOPLEVEL_CACHE.clear()
|
||
self.assertNotEqual(pmod.slug(a), pmod.slug(b))
|
||
|
||
|
||
class TestTranscriptScope(unittest.TestCase):
|
||
"""Regression: scanning every project against the current dictionary reported 802
|
||
leaks, nearly all from unrelated sessions. The one metric the design leans on has
|
||
to be scoped to mean anything."""
|
||
|
||
def test_project_dir_encoding(self):
|
||
from ctxguard.transcript import project_transcript_dir
|
||
got = project_transcript_dir("/home/u/Work/sandbox/demo-service").name
|
||
self.assertEqual(got, "-home-u-Work-sandbox-demo-service")
|
||
|
||
def test_missing_project_dir_yields_nothing(self):
|
||
from ctxguard.transcript import find_transcripts
|
||
self.assertEqual(find_transcripts(cwd="/nonexistent/xyz-project"), [])
|
||
|
||
def test_all_projects_is_opt_in(self):
|
||
from ctxguard.cli import build_parser
|
||
args = build_parser().parse_args(["scan-transcript"])
|
||
self.assertFalse(args.all_projects)
|
||
|
||
|
||
class TestHelpers(unittest.TestCase):
|
||
def test_base26(self):
|
||
self.assertEqual([base26(n) for n in (1, 26, 27, 52)], ["A", "Z", "AA", "AZ"])
|
||
|
||
def test_luhn(self):
|
||
self.assertTrue(detect.luhn_ok("4111111111111111"))
|
||
self.assertFalse(detect.luhn_ok("4111111111111112"))
|
||
|
||
def test_withheld_carries_no_value(self):
|
||
f = detect.Finding("aws-access-key", "secret", 0, 20, "AKIA1234567890ABCDEF", "<x>")
|
||
self.assertNotIn("AKIA1234567890ABCDEF", repr(detect.withheld(f)))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|