From b4a6dc2f4fa1e2520f0ad224c042a1e17dc4efe5 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 13:48:25 +0200 Subject: [PATCH 1/2] test: define configuration-file secrets and analysis coverage --- tests/test_cli.py | 5 +- tests/test_config_secrets_coverage.py | 229 ++++++++++++++++++++++++++ tests/test_detectors.py | 1 + tests/test_discovery.py | 4 +- tests/test_project.py | 1 + tests/test_syntax_coverage.py | 2 +- 6 files changed, 237 insertions(+), 5 deletions(-) create mode 100644 tests/test_config_secrets_coverage.py diff --git a/tests/test_cli.py b/tests/test_cli.py index 996e58d..794c81d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -46,6 +46,7 @@ def test_check_reports_findings_from_loaded_plugins(tmp_path, capsys) -> None: f"{source}:2:5: high dangerous-eval: call to python.builtins.eval executes" " dynamically built code [run]\n" "1 finding\n" + "coverage: 1/1 files, 1/1 functions\n" ) @@ -53,14 +54,14 @@ def test_check_is_clean_when_nothing_is_found(tmp_path, capsys) -> None: source = write(tmp_path, "def run(code):\n return code\n") assert main(["--check", str(source), "--plugins", str(PLUGINS)]) == 0 - assert capsys.readouterr().out == "no findings\n" + assert capsys.readouterr().out == "no findings\ncoverage: 1/1 files, 1/1 functions\n" def test_check_without_plugins_finds_nothing(tmp_path, capsys) -> None: source = write(tmp_path, "def run(code):\n eval(code)\n") assert main(["--check", str(source)]) == 0 - assert capsys.readouterr().out == "no findings\n" + assert capsys.readouterr().out == "no findings\ncoverage: 1/1 files, 1/1 functions\n" def test_check_json_format(tmp_path, capsys) -> None: diff --git a/tests/test_config_secrets_coverage.py b/tests/test_config_secrets_coverage.py new file mode 100644 index 0000000..750508f --- /dev/null +++ b/tests/test_config_secrets_coverage.py @@ -0,0 +1,229 @@ +"""Acceptance tests for two remaining audit recommendations: secrets in configuration +files and a coverage metric that tells "nothing found" from "nothing analysed". + +- ``config-secrets`` is a project plugin walking ``.env``, YAML, TOML, JSON, INI and + properties files under the root, decoded by their byte order mark, and judging each + key-value pair with the same provider patterns, credential names and entropy rule as + the Python literal detector; virtual environments and tooling directories are skipped. +- ``Coverage`` records, per file, whether it was analysed or why not, and how many of + its functions were; the text and JSON reporters print it when a report carries one. + +Expected to remain red until ``findings.coverage``, ``ProjectContext.root``, +``engine.analyze_file`` and the ``config-secrets`` plugin exist. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.reporters import Report, render_json, render_text +from coretrace_python.source import SourceId, SourceManager, SourceSpan + +try: + from coretrace_python.findings.coverage import Coverage, FileCoverage + from coretrace_python.plugins import ProjectContext + from coretrace_python.plugins.secrets import config_literals +except ImportError as error: # pragma: no cover - red until the pass lands + MISSING: Exception | None = error +else: + MISSING = None + if "root" not in ProjectContext.__init__.__code__.co_varnames or not hasattr(engine, "analyze_file"): + MISSING = AttributeError("ProjectContext.root or engine.analyze_file is missing") + + +@pytest.fixture(autouse=True) +def require_pass() -> None: + if MISSING is not None: + pytest.fail(f"config secrets and coverage are not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" +GITHUB_TOKEN = "ghp_" + "a1B2c3D4e5F6g7H8i9J0k1L2m3N4o5P6q7R8" + + +def project(root: Path, files: dict[str, str | bytes]) -> Path: + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + if isinstance(text, bytes): + path.write_bytes(text) + else: + path.write_text(text, encoding="utf-8") + return root + + +def located(findings: tuple[Finding, ...]) -> list[tuple[str, int, str, str]]: + return sorted( + (Path(str(f.span.source_id)).name, f.span.start_line, f.rule_id, f.metadata.get("name", "")) + for f in findings + ) + + +# --------------------------------------------------------------------------- config literals + + +def test_config_files_yield_key_value_literals(tmp_path: Path) -> None: + root = project( + tmp_path, + { + ".env": "DB_PASSWORD=hunter2\nDEBUG=true\nEMPTY=\nQUOTED='with space'\n", + "config.yaml": "service:\n token: abc\n - name: item\n", + "settings.toml": '[db]\npassword = "s3cr3t"\nport = 5432\n', + "insomnia.json": '{"resources": [{"token": "xyz", "count": 3}]}', + "app.ini": "[main]\napi_key = k-1234\n", + "env/pyvenv.cfg": "", + "env/Lib/site.cfg": "password = nope\n", + "app.py": "PASSWORD = 'py'\n", + }, + ) + + found = sorted((Path(str(span.source_id)).name, span.start_line, name, value) for value, name, span, _ in config_literals(root)) + + assert found == [ + (".env", 1, "DB_PASSWORD", "hunter2"), + (".env", 2, "DEBUG", "true"), + (".env", 4, "QUOTED", "with space"), + ("app.ini", 2, "api_key", "k-1234"), + ("config.yaml", 2, "token", "abc"), + ("config.yaml", 3, "name", "item"), + ("insomnia.json", 1, "token", "xyz"), + ("settings.toml", 2, "password", "s3cr3t"), + ] + + +def test_config_files_honour_their_byte_order_mark(tmp_path: Path) -> None: + root = project(tmp_path, {"export.yaml": "token: abc\n".encode("utf-16"), "app.py": "x = 1\n"}) + assert [(name, value) for value, name, _, _ in config_literals(root)] == [("token", "abc")] + + +# --------------------------------------------------------------------------- shipped plugin + + +def test_config_secrets_are_reported_with_the_python_rules(tmp_path: Path) -> None: + root = project( + tmp_path, + { + ".env": "DB_PASSWORD=hunter2\nDEBUG=true\nSECRET_KEY=${SECRET_KEY}\n", + "config.yaml": f"github:\n token: {GITHUB_TOKEN}\n", + "insomnia.json": '{"environment": {"TOKEN": "ZmFrZS10b2tlbi12YWx1ZS1mb3ItdGVzdGluZy0xMjM0NTY3ODkw"}}', + "app.py": "x = 1\n", + }, + ) + + findings = engine.analyze_project(root, [PLUGINS]).findings + + assert located(findings) == [ + (".env", 1, "hardcoded-credential", "DB_PASSWORD"), + ("config.yaml", 2, "hardcoded-secret", "token"), + ("insomnia.json", 1, "hardcoded-credential", "TOKEN"), + ] + assert all(f.function is None for f in findings) + assert all(GITHUB_TOKEN not in f.message and "hunter2" not in f.message for f in findings) + assert {f.confidence for f in findings if f.rule_id == "hardcoded-secret"} == {Confidence.HIGH} + + +def test_python_files_are_left_to_the_module_detector(tmp_path: Path) -> None: + root = project(tmp_path, {"settings.py": "PASSWORD = 'hunter2'\n"}) + findings = engine.analyze_project(root, [PLUGINS]).findings + assert [(f.rule_id, f.metadata["name"]) for f in findings] == [("hardcoded-credential", "PASSWORD")] + + +def test_project_context_exposes_the_root(tmp_path: Path) -> None: + root = project(tmp_path, {"app.py": "x = 1\n"}) + from coretrace_python.plugins import ProjectPlugin + + seen: list[Path] = [] + + class Peek(ProjectPlugin): + name = "peek" + + def analyze_project(self, ctx): # type: ignore[no-untyped-def] + seen.append(ctx.root) + return () + + engine.analyze_project(root, plugins=[Peek()]) + assert seen == [root] + + +# --------------------------------------------------------------------------- coverage + + +def test_coverage_counts_files_and_functions() -> None: + coverage = Coverage( + ( + FileCoverage("a.py", "analysed", 3, 2), + FileCoverage("b.py", "syntax-error", 0, 0), + FileCoverage("c.py", "unreadable", 0, 0), + FileCoverage("d.py", "analysed", 1, 1), + ) + ) + + assert (coverage.files, coverage.files_analysed) == (4, 2) + assert (coverage.functions, coverage.functions_analysed) == (4, 3) + assert coverage.summary() == "coverage: 2/4 files, 3/4 functions" + + +def test_project_coverage_reflects_notes(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "clean.py": "def a():\n return 1\n\ndef b():\n return 2\n", + "partial.py": "def ok():\n return 1\n\ndef odd():\n class Inner:\n pass\n return Inner\n", + "broken.py": "def (:\n", + "legacy.py": b"\xff\xfeprint('x')\n", + }, + ) + + coverage = engine.analyze_project(root, [PLUGINS]).coverage + + assert {(c.path.split("/")[-1], c.status, c.functions, c.analysed) for c in coverage.details} == { + ("clean.py", "analysed", 2, 2), + ("partial.py", "analysed", 2, 1), + ("broken.py", "syntax-error", 0, 0), + ("legacy.py", "unreadable", 0, 0), + } + assert coverage.summary() == "coverage: 2/4 files, 3/4 functions" + + +def test_single_file_analysis_carries_coverage() -> None: + analysis = engine.analyze_file( + SourceManager().add_source("one.py", "def a():\n return 1\n\ndef b():\n class C:\n pass\n"), [PLUGINS] + ) + assert analysis.coverage.summary() == "coverage: 1/1 files, 1/2 functions" + assert [f.rule_id for f in analysis.findings] == ["unsupported-syntax"] + + +def test_reporters_print_coverage_when_present() -> None: + finding = Finding("r", "m", Severity.LOW, Confidence.HIGH, SourceSpan(SourceId("a.py"), 1, 1)) + coverage = Coverage((FileCoverage("a.py", "analysed", 2, 2),)) + + assert render_text(Report((finding,), "t", "1")) == "a.py:1:1: low r: m\n1 finding\n" + assert render_text(Report((finding,), "t", "1", coverage)) == ( + "a.py:1:1: low r: m\n1 finding\ncoverage: 1/1 files, 2/2 functions\n" + ) + document = json.loads(render_json(Report((), "t", "1", coverage))) + assert document["coverage"] == { + "files": 1, + "files_analysed": 1, + "functions": 2, + "functions_analysed": 2, + "details": [{"path": "a.py", "status": "analysed", "functions": 2, "analysed": 2}], + } + assert "coverage" not in json.loads(render_json(Report((), "t", "1"))) + + +def test_cli_prints_coverage_for_checks(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + source = tmp_path / "one.py" + source.write_text("def a():\n return 1\n", encoding="utf-8") + + assert main(["--check", str(source), "--plugins", str(PLUGINS)]) == 0 + assert capsys.readouterr().out == "no findings\ncoverage: 1/1 files, 1/1 functions\n" + assert main(["--check", str(tmp_path), "--plugins", str(PLUGINS), "--format", "json"]) == 0 + assert json.loads(capsys.readouterr().out)["coverage"]["files"] == 1 diff --git a/tests/test_detectors.py b/tests/test_detectors.py index aa41e9d..8deac74 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -163,6 +163,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: assert set(by_name) == { "command-injection", + "config-secrets", "dangerous-eval", "dependency-policy", "django-models", diff --git a/tests/test_discovery.py b/tests/test_discovery.py index 28e8699..c702671 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -75,7 +75,7 @@ def test_dashed_packages_are_analysed_instead_of_crashing(tmp_path: Path) -> Non def test_undecodable_files_become_notes_and_the_rest_is_analysed(tmp_path: Path) -> None: root = project(tmp_path, {"app.py": "import os\n\ndef run():\n os.system(input())\n"}) - (root / "legacy.py").write_bytes(b"\xff\xfeprint('utf-16')\n") + (root / "legacy.py").write_bytes(b"print('x')\n\xff") analysis = engine.analyze_project(root, [Path(__file__).resolve().parent.parent / "plugins"]) @@ -83,7 +83,7 @@ def test_undecodable_files_become_notes_and_the_rest_is_analysed(tmp_path: Path) assert ("syntax-error", "legacy.py") in notes assert ("command-injection", "app.py") in notes assert set(analysis.keys) == {"app"} - assert any("utf-8" in f.message for f in analysis.findings if f.rule_id == "syntax-error") + assert any("codec" in f.message for f in analysis.findings if f.rule_id == "syntax-error") def test_utf16_dependency_files_are_read(tmp_path: Path) -> None: diff --git a/tests/test_project.py b/tests/test_project.py index 8a5cab0..7b1d44f 100644 --- a/tests/test_project.py +++ b/tests/test_project.py @@ -260,6 +260,7 @@ def test_check_accepts_a_directory(tmp_path: Path, capsys) -> None: # type: ign f"{(root / 'app' / 'main.py').resolve()}:4:5: high command-injection: Command injection:" " stdin input reaches python.os.system through app.helpers.execute [run]\n" "1 finding\n" + "coverage: 4/4 files, 3/3 functions\n" ) diff --git a/tests/test_syntax_coverage.py b/tests/test_syntax_coverage.py index 442394a..92efd60 100644 --- a/tests/test_syntax_coverage.py +++ b/tests/test_syntax_coverage.py @@ -282,7 +282,7 @@ def test_check_reports_unsupported_functions_and_keeps_going(tmp_path, capsys) - assert f"{source}:3:1: info unsupported-syntax: " in output assert "Class" in output assert f"{source}:10:9: high dangerous-eval:" in output - assert output.endswith("2 findings\n") + assert output.endswith("2 findings\ncoverage: 1/1 files, 1/2 functions\n") def test_unsupported_syntax_findings_are_notes() -> None: From a66c6c1569caf69e40129358c9c04931c4ef654d Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 13:48:26 +0200 Subject: [PATCH 2/2] feat: scan configuration files for secrets and report analysis coverage --- README.md | 11 +- .../secrets/config_secrets/config_secrets.py | 38 ++++++ plugins/secrets/config_secrets/plugin.toml | 9 ++ .../hardcoded_secrets/hardcoded_secrets.py | 42 ++---- src/coretrace_python/cli.py | 6 +- src/coretrace_python/engine.py | 56 +++++--- src/coretrace_python/findings/__init__.py | 10 +- src/coretrace_python/findings/coverage.py | 45 +++++++ .../interprocedural/__init__.py | 2 + .../interprocedural/modulegraph.py | 30 +++-- src/coretrace_python/plugins/api.py | 3 + src/coretrace_python/plugins/secrets.py | 120 +++++++++++++++++- src/coretrace_python/reporters/json_format.py | 12 ++ src/coretrace_python/reporters/report.py | 3 +- src/coretrace_python/reporters/text.py | 2 + src/coretrace_python/source/__init__.py | 10 +- src/coretrace_python/source/manager.py | 13 +- 17 files changed, 341 insertions(+), 71 deletions(-) create mode 100644 plugins/secrets/config_secrets/config_secrets.py create mode 100644 plugins/secrets/config_secrets/plugin.toml create mode 100644 src/coretrace_python/findings/coverage.py diff --git a/README.md b/README.md index 81bb055..0c6e3d6 100644 --- a/README.md +++ b/README.md @@ -71,9 +71,10 @@ keys, JWTs, SendGrid, Twilio), for credential-like names bound to a real value (`password`, `token`, `api_key` and the like, bound by name, attribute or constant key such as `app.config['SECRET_KEY']`, placeholders excluded) and for opaque high-entropy tokens; one finding per literal, at high, medium and low confidence -respectively, with a redacted preview and never the secret itself. Other plugins add -providers by subclassing `SecretDetector` with their own patterns. Configuration files -are not scanned. +respectively, with a redacted preview and never the secret itself. The `config-secrets` +plugin applies the same rules to the key-value pairs of `.env`, YAML, TOML, JSON, INI and +properties files under the project root. Other plugins add providers by subclassing +`SecretDetector` with their own patterns. `--cache DIR` keeps the results of a directory check on disk, one JSON entry per module. The entry is keyed by the module's source, the engine and plugin versions, the plugin @@ -121,7 +122,9 @@ a helper that fills a list taints the caller's list, across files too. Each flow is then judged: a dominating guard that proves the value safe (`isdigit()`, membership in a constant allowlist, equality with a constant) refutes it and it is not reported; a guard that only mentions the value makes it a hotspot with medium confidence; -the verdict and its evidence are in the finding metadata. Three more kinds of evidence +the verdict and its evidence are in the finding metadata. Every check ends with a +coverage line, `coverage: 3/4 files, 5/6 functions`, and the JSON report carries the +per-file detail, so "no findings" can be told from "nothing analysed". Three more kinds of evidence apply: a value proven numeric by the range analysis (`int()`, `len()`, arithmetic, bounded by the comparisons on the path) cannot inject; a `Validator` model names a callable whose truth proves one of its arguments, such as `re.fullmatch`; and an diff --git a/plugins/secrets/config_secrets/config_secrets.py b/plugins/secrets/config_secrets/config_secrets.py new file mode 100644 index 0000000..1b8374e --- /dev/null +++ b/plugins/secrets/config_secrets/config_secrets.py @@ -0,0 +1,38 @@ +"""Hardcoded secrets in configuration files: ``.env``, YAML, TOML, JSON, INI, properties. + +Judged with the same rules as the Python literal detector, once per project. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import ClassVar + +from coretrace_python.findings import Finding +from coretrace_python.plugins import PluginContext, ProjectContext, ProjectPlugin +from coretrace_python.plugins.secrets import ( + DEFAULT_CREDENTIAL_NAMES, + DEFAULT_PATTERNS, + SecretDetector, + SecretPattern, + config_literals, +) + + +class ConfigSecrets(SecretDetector, ProjectPlugin): + name: ClassVar[str] = "config-secrets" + patterns: ClassVar[tuple[SecretPattern, ...]] = DEFAULT_PATTERNS + credential_names: ClassVar[tuple[str, ...]] = DEFAULT_CREDENTIAL_NAMES + + def analyze(self, ctx: PluginContext) -> Sequence[Finding]: + return () + + def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: + if ctx.root is None: + return () + findings: list[Finding] = [] + for value, name, span, function in config_literals(ctx.root): + finding = self.judge(value, name, span, function) + if finding is not None: + findings.append(finding) + return findings diff --git a/plugins/secrets/config_secrets/plugin.toml b/plugins/secrets/config_secrets/plugin.toml new file mode 100644 index 0000000..c848bee --- /dev/null +++ b/plugins/secrets/config_secrets/plugin.toml @@ -0,0 +1,9 @@ +name = "config-secrets" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["secret.config-files"] + +[entrypoint] +module = "config_secrets" +class = "ConfigSecrets" diff --git a/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py b/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py index ae87981..7c9e1f2 100644 --- a/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py +++ b/plugins/secrets/hardcoded_secrets/hardcoded_secrets.py @@ -1,41 +1,19 @@ -"""Hardcoded secrets: provider formats, credential-like names, high-entropy tokens.""" +"""Hardcoded secrets in Python literals: provider formats, credential-like names, +high-entropy tokens.""" from __future__ import annotations from typing import ClassVar -from coretrace_python.plugins.secrets import SecretDetector, SecretPattern +from coretrace_python.plugins.secrets import ( + DEFAULT_CREDENTIAL_NAMES, + DEFAULT_PATTERNS, + SecretDetector, + SecretPattern, +) class HardcodedSecrets(SecretDetector): name: ClassVar[str] = "hardcoded-secrets" - patterns: ClassVar[tuple[SecretPattern, ...]] = ( - SecretPattern("aws", r"\b(AKIA|ASIA)[0-9A-Z]{16}\b"), - SecretPattern("github", r"\bgh[pousr]_[A-Za-z0-9]{36,}\b"), - SecretPattern("github", r"\bgithub_pat_[A-Za-z0-9]{22}_[A-Za-z0-9]{59}\b"), - SecretPattern("slack", r"\bxox[abpr]-[0-9]{10,}-[0-9A-Za-z-]{10,}"), - SecretPattern("stripe", r"\b[sr]k_(live|test)_[0-9A-Za-z]{24,}\b"), - SecretPattern("google", r"\bAIza[0-9A-Za-z_-]{35}\b"), - SecretPattern("private-key", r"-----BEGIN [A-Z ]*PRIVATE KEY-----"), - SecretPattern("jwt", r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b"), - SecretPattern("sendgrid", r"\bSG\.[A-Za-z0-9_-]{22}\.[A-Za-z0-9_-]{43}\b"), - SecretPattern("twilio", r"\bSK[0-9a-fA-F]{32}\b"), - SecretPattern( - "url", - r"://[^/\s:@'\"]+:[^/\s@'\"]+@" - r"|[?&](?:password|passwd|pwd|token|api_key|apikey|secret|access_key)=[^&\s'\"]{3,}", - ), - ) - credential_names: ClassVar[tuple[str, ...]] = ( - "password", - "passwd", - "pwd", - "secret", - "token", - "api_key", - "apikey", - "api-key", - "private_key", - "access_key", - "credential", - ) + patterns: ClassVar[tuple[SecretPattern, ...]] = DEFAULT_PATTERNS + credential_names: ClassVar[tuple[str, ...]] = DEFAULT_CREDENTIAL_NAMES diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index ffb265f..ff2d1cf 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -184,14 +184,16 @@ def main(argv: list[str] | None = None) -> int: policy_file=args.policy, ) findings = analysis.findings + coverage = analysis.coverage if args.sbom is not None: args.sbom.write_text( render_sbom(analysis.dependencies, analysis.advisories, engine.TOOL_NAME, __version__), encoding="utf-8", ) else: - findings = engine.check(SourceManager().load_file(args.path), args.plugins) - print(render(args.format or "text", engine.report(findings)), end="") + file_analysis = engine.analyze_file(SourceManager().load_file(args.path), args.plugins) + findings, coverage = file_analysis.findings, file_analysis.coverage + print(render(args.format or "text", engine.report(findings, coverage)), end="") return EXIT_FINDINGS if findings else EXIT_CLEAN except _ANALYSIS_ERRORS as error: print(f"error: {error}", file=sys.stderr) diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index efd413c..d3b41a7 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -49,7 +49,14 @@ parse_dependencies, ) from coretrace_python.dependency.correlation import advisory_sinks, affected_symbols, correlate -from coretrace_python.findings import FINDING_SCHEMA_VERSION, Confidence, Finding, Severity +from coretrace_python.findings import ( + FINDING_SCHEMA_VERSION, + Confidence, + Coverage, + FileCoverage, + Finding, + Severity, +) from coretrace_python.findings.refutation import RefutationAnalysis from coretrace_python.frontend import HIRBuildError, ParseError, build_hir from coretrace_python.hir import HIR_SCHEMA_VERSION, nodes @@ -89,7 +96,7 @@ from coretrace_python.semantic.imports import ImportAnalysis, ImportResolutionError, ImportTable from coretrace_python.semantic.scopes import ScopeError from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.source import SourceFile, SourceManager, SourceSpan +from coretrace_python.source import SourceFile, SourceManager, SourceSpan, decode_text from coretrace_python.taint import ( ModelTable, SecurityModelAnalysis, @@ -171,6 +178,13 @@ class ProjectAnalysis: keys: Mapping[str, str] = field(default_factory=_no_keys) reused: tuple[str, ...] = () advisories: tuple[Advisory, ...] = () + coverage: Coverage = field(default_factory=Coverage) + + +@dataclass(frozen=True) +class FileAnalysis: + findings: tuple[Finding, ...] + coverage: Coverage def build_manager( @@ -202,11 +216,20 @@ def plugin_models(plugins: Iterable[Plugin]) -> ModelTable: def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ...]: """Run every plugin found under ``plugin_roots`` against one file.""" + return analyze_file(source, plugin_roots).findings + + +def analyze_file(source: SourceFile, plugin_roots: Sequence[Path]) -> FileAnalysis: + """One file's findings and coverage, with every plugin found under ``plugin_roots``.""" + manager = _register_all(build_hir(source)) registry = load_plugins(plugin_roots, manager) manager.provide(SecurityModelAnalysis, plugin_models(loaded.plugin for loaded in registry)) manager.provide(ProjectSummaries, SummaryIndex()) - return _check_module(manager, tuple(loaded.plugin for loaded in registry))[0] + findings, supported = _check_module(manager, tuple(loaded.plugin for loaded in registry)) + functions = len(analyzable_functions(manager.module)) + coverage = Coverage((FileCoverage(str(source.source_id), "analysed", functions, len(supported)),)) + return FileAnalysis(findings, coverage) def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph: @@ -218,7 +241,7 @@ def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph: if not path.is_file(): continue try: - text = _decode_manifest(path.read_bytes()) + text = decode_text(path.read_bytes()) except (OSError, UnicodeDecodeError) as error: graph = graph.merge(DependencyGraph(errors=(f"{path}: {error}",))) continue @@ -226,16 +249,6 @@ def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph: return graph -def _decode_manifest(data: bytes) -> str: - """Dependency files written by ``pip freeze`` on Windows are UTF-16 with a byte - order mark; honour the mark, otherwise expect UTF-8.""" - - for mark, encoding in ((b"\xef\xbb\xbf", "utf-8-sig"), (b"\xff\xfe", "utf-16"), (b"\xfe\xff", "utf-16")): - if data.startswith(mark): - return data.decode(encoding) - return data.decode("utf-8") - - def analyze_project( root: Path, plugin_roots: Sequence[Path] = (), @@ -278,14 +291,17 @@ def analyze_project( modules: dict[str, nodes.Module] = {} files: dict[str, SourceFile] = {} unreadable: list[tuple[Path, str]] = [] + coverage: list[FileCoverage] = [] discovered = discover_sources(root, sources, unreadable) for path, reason in unreadable: findings.append(_note("syntax-error", f"{path}: {reason}", sources.add_source(str(path), ""), 1)) + coverage.append(FileCoverage(str(path), "unreadable", 0, 0)) for source in discovered: try: modules[source.module_name] = build_hir(source) except (ParseError, HIRBuildError) as error: findings.append(_note("syntax-error", str(error), source, 1)) + coverage.append(FileCoverage(str(source.source_id), "syntax-error", 0, 0)) continue files[source.module_name] = source @@ -309,6 +325,7 @@ def analyze_project( imports[name] = manager.get(ImportAnalysis) except (ImportResolutionError, ScopeError) as error: findings.append(_note("syntax-error", str(error), files[name], 1)) + coverage.append(FileCoverage(str(files[name].source_id), "syntax-error", 0, 0)) continue analysable[name] = manager graph = build_module_graph( @@ -374,11 +391,15 @@ def analyze_project( for name in sorted(analysable): entry = results[name] findings.extend(entry.findings) + unsupported = sum(1 for f in entry.findings if f.rule_id == "unsupported-syntax") + coverage.append( + FileCoverage(str(files[name].source_id), "analysed", len(entry.functions), len(entry.functions) - unsupported) + ) sites: dict[str, list[CallSite]] = {function: [] for function in entry.functions} for site in entry.sites: sites.setdefault(site.caller, []).append(site) call_graphs[name] = CallGraph({}, {f: tuple(s) for f, s in sites.items()}, frozenset()) - context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs, policy) + context = ProjectContext(graph, dependencies, advisories, analysable, call_graphs, policy, root) for plugin in all_plugins: if isinstance(plugin, ProjectPlugin): findings.extend(plugin.analyze_project(context)) @@ -390,6 +411,7 @@ def analyze_project( MappingProxyType(keys), reused, advisories, + Coverage(tuple(sorted(coverage, key=lambda c: c.path))), ) @@ -612,5 +634,5 @@ def _register_all(module: nodes.Module) -> AnalysisManager: return manager -def report(findings: Sequence[Finding]) -> Report: - return Report(tuple(findings), TOOL_NAME, __version__) +def report(findings: Sequence[Finding], coverage: Coverage | None = None) -> Report: + return Report(tuple(findings), TOOL_NAME, __version__, coverage) diff --git a/src/coretrace_python/findings/__init__.py b/src/coretrace_python/findings/__init__.py index 4692b91..62b859e 100644 --- a/src/coretrace_python/findings/__init__.py +++ b/src/coretrace_python/findings/__init__.py @@ -1,5 +1,13 @@ """Normalized findings produced by plugins and consumed by reporters.""" +from coretrace_python.findings.coverage import Coverage, FileCoverage from coretrace_python.findings.model import FINDING_SCHEMA_VERSION, Confidence, Finding, Severity -__all__ = ["FINDING_SCHEMA_VERSION", "Confidence", "Finding", "Severity"] +__all__ = [ + "FINDING_SCHEMA_VERSION", + "Confidence", + "Coverage", + "FileCoverage", + "Finding", + "Severity", +] diff --git a/src/coretrace_python/findings/coverage.py b/src/coretrace_python/findings/coverage.py new file mode 100644 index 0000000..717ac4c --- /dev/null +++ b/src/coretrace_python/findings/coverage.py @@ -0,0 +1,45 @@ +"""Analysis coverage: which files and functions a run actually looked at. + +"No findings" only means something when the reader knows what was analysed. Each file +is ``analysed``, a ``syntax-error`` (the frontend rejected it) or ``unreadable`` (it could +not be decoded); analysed files count their functions and how many lowered. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class FileCoverage: + path: str + status: str + functions: int + analysed: int + + +@dataclass(frozen=True) +class Coverage: + details: tuple[FileCoverage, ...] = () + + @property + def files(self) -> int: + return len(self.details) + + @property + def files_analysed(self) -> int: + return sum(1 for d in self.details if d.status == "analysed") + + @property + def functions(self) -> int: + return sum(d.functions for d in self.details) + + @property + def functions_analysed(self) -> int: + return sum(d.analysed for d in self.details) + + def summary(self) -> str: + return ( + f"coverage: {self.files_analysed}/{self.files} files, " + f"{self.functions_analysed}/{self.functions} functions" + ) diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index 53838ca..f53b497 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -12,6 +12,7 @@ from coretrace_python.interprocedural.modulegraph import ( ModuleGraph, build_module_graph, + discover_files, discover_sources, project_symbol, ) @@ -42,6 +43,7 @@ "Target", "UnknownTarget", "build_module_graph", + "discover_files", "discover_sources", "project_symbol", ] diff --git a/src/coretrace_python/interprocedural/modulegraph.py b/src/coretrace_python/interprocedural/modulegraph.py index ee357ff..2360369 100644 --- a/src/coretrace_python/interprocedural/modulegraph.py +++ b/src/coretrace_python/interprocedural/modulegraph.py @@ -24,22 +24,34 @@ IGNORED_DIRECTORIES = frozenset({"__pycache__", "node_modules", "venv", "build", "dist", "site-packages"}) -def discover_sources( - root: Path, manager: SourceManager, errors: list[tuple[Path, str]] | None = None -) -> tuple[SourceFile, ...]: - """Load every ``.py`` file under ``root``, skipping hidden and tooling directories and - virtual environments, recognised by the ``pyvenv.cfg`` at their root whatever their - name. A file that cannot be read or decoded is skipped and reported in ``errors``: - Python could not import it either.""" +def discover_files(root: Path, pattern: str = "*") -> tuple[Path, ...]: + """Every file matching ``pattern`` under ``root``, skipping hidden and tooling + directories and virtual environments, recognised by the ``pyvenv.cfg`` at their root + whatever their name.""" - found: list[SourceFile] = [] + found: list[Path] = [] environments: dict[Path, bool] = {} - for path in sorted(root.rglob("*.py")): + for path in sorted(root.rglob(pattern)): + if not path.is_file(): + continue relative = path.relative_to(root) if any(p.startswith(".") or p in IGNORED_DIRECTORIES for p in relative.parts[:-1]): continue if any(_is_environment(root / Path(*relative.parts[:depth]), environments) for depth in range(1, len(relative.parts))): continue + found.append(path) + return tuple(found) + + +def discover_sources( + root: Path, manager: SourceManager, errors: list[tuple[Path, str]] | None = None +) -> tuple[SourceFile, ...]: + """Load every ``.py`` file under ``root`` (see ``discover_files``). A file that cannot + be read or decoded is skipped and reported in ``errors``: Python could not import it + either.""" + + found: list[SourceFile] = [] + for path in discover_files(root, "*.py"): try: found.append(manager.load_file(path)) except (OSError, UnicodeDecodeError) as error: diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py index 5d20413..0566e66 100644 --- a/src/coretrace_python/plugins/api.py +++ b/src/coretrace_python/plugins/api.py @@ -10,6 +10,7 @@ from abc import ABC, abstractmethod from collections.abc import Iterable, Mapping, Sequence +from pathlib import Path from typing import Any, ClassVar, overload from coretrace_python.analysis import ( @@ -66,7 +67,9 @@ def __init__( managers: Mapping[str, AnalysisManager], call_graphs: Mapping[str, CallGraph] | None = None, policy: Policy | None = None, + root: Path | None = None, ) -> None: + self.root = root self.graph = graph self.dependencies = dependencies self.advisories = advisories diff --git a/src/coretrace_python/plugins/secrets.py b/src/coretrace_python/plugins/secrets.py index 084c856..31fe8b4 100644 --- a/src/coretrace_python/plugins/secrets.py +++ b/src/coretrace_python/plugins/secrets.py @@ -11,18 +11,22 @@ from __future__ import annotations +import json import math import re +import tomllib from collections import Counter from collections.abc import Iterator, Sequence from dataclasses import dataclass +from pathlib import Path from typing import ClassVar from coretrace_python.findings import Confidence, Finding, Severity from coretrace_python.hir import nodes from coretrace_python.hir.visitors import Node, children +from coretrace_python.interprocedural import discover_files from coretrace_python.plugins.api import Plugin, PluginContext -from coretrace_python.source import SourceSpan +from coretrace_python.source import SourceId, SourceSpan, decode_text Literal = tuple[str, str | None, SourceSpan, str | None] @@ -133,6 +137,120 @@ def matches(self, text: str) -> bool: return re.search(self.regex, text) is not None +DEFAULT_PATTERNS: tuple[SecretPattern, ...] = ( + SecretPattern("aws", r"\b(AKIA|ASIA)[0-9A-Z]{16}\b"), + SecretPattern("github", r"\bgh[pousr]_[A-Za-z0-9]{36,}\b"), + SecretPattern("github", r"\bgithub_pat_[A-Za-z0-9]{22}_[A-Za-z0-9]{59}\b"), + SecretPattern("slack", r"\bxox[abpr]-[0-9]{10,}-[0-9A-Za-z-]{10,}"), + SecretPattern("stripe", r"\b[sr]k_(live|test)_[0-9A-Za-z]{24,}\b"), + SecretPattern("google", r"\bAIza[0-9A-Za-z_-]{35}\b"), + SecretPattern("private-key", r"-----BEGIN [A-Z ]*PRIVATE KEY-----"), + SecretPattern("jwt", r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b"), + SecretPattern("sendgrid", r"\bSG\.[A-Za-z0-9_-]{22}\.[A-Za-z0-9_-]{43}\b"), + SecretPattern("twilio", r"\bSK[0-9a-fA-F]{32}\b"), + SecretPattern( + "url", + r"://[^/\s:@'\"]+:[^/\s@'\"]+@" + r"|[?&](?:password|passwd|pwd|token|api_key|apikey|secret|access_key)=[^&\s'\"]{3,}", + ), +) + +DEFAULT_CREDENTIAL_NAMES: tuple[str, ...] = ( + "password", + "passwd", + "pwd", + "secret", + "token", + "api_key", + "apikey", + "api-key", + "private_key", + "access_key", + "credential", +) + +CONFIG_SUFFIXES = frozenset({".env", ".yaml", ".yml", ".toml", ".json", ".ini", ".cfg", ".properties", ".conf"}) +_PAIR = re.compile(r"^\s*(?:-\s+)?([A-Za-z_][\w.-]*)\s*[:=]\s*(.*?)\s*$") +_MAX_CONFIG_BYTES = 8_000_000 + + +def config_literals(root: Path) -> Iterator[Literal]: + """Every string value of the configuration files under ``root`` with the key it is + bound to: ``.env``, YAML, TOML, JSON, INI and properties files, decoded by their + byte order mark. Python files are left to ``literals``.""" + + for path in discover_files(root): + if not (path.suffix in CONFIG_SUFFIXES or path.name.startswith(".env")): + continue + try: + data = path.read_bytes() + if len(data) > _MAX_CONFIG_BYTES: + continue + text = decode_text(data) + except (OSError, UnicodeDecodeError): + continue + source = SourceId(str(path)) + if path.suffix == ".json": + yield from _structured(source, text, _load_json(text)) + elif path.suffix == ".toml": + yield from _structured(source, text, _load_toml(text)) + else: + yield from _pairs(source, text) + + +def _load_json(text: str) -> object: + try: + return json.loads(text) + except ValueError: + return None + + +def _load_toml(text: str) -> object: + try: + return tomllib.loads(text) + except tomllib.TOMLDecodeError: + return None + + +def _structured(source: SourceId, text: str, data: object) -> Iterator[Literal]: + lines = text.splitlines() + + def line_of(key: str) -> int: + for number, line in enumerate(lines, start=1): + if re.match(rf'^\s*"?{re.escape(key)}"?\s*[:=]', line) or f'"{key}"' in line: + return number + return 1 + + def walk(node: object, key: str | None) -> Iterator[Literal]: + if isinstance(node, dict): + for name, value in node.items(): + yield from walk(value, str(name)) + elif isinstance(node, list): + for item in node: + yield from walk(item, key) + elif isinstance(node, str) and key is not None: + yield node, key, SourceSpan(source, line_of(key), 1), None + + yield from walk(data, None) + + +def _pairs(source: SourceId, text: str) -> Iterator[Literal]: + for number, line in enumerate(text.splitlines(), start=1): + stripped = line.strip() + if not stripped or stripped[0] in "#;[": + continue + match = _PAIR.match(line) + if match is None: + continue + key, value = match.group(1), match.group(2) + if value.startswith("#"): + continue + if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"": + value = value[1:-1] + if value: + yield value, key, SourceSpan(source, number, 1), None + + def redacted(value: str) -> str: preview = value[:4] if len(value) > 8 else value[:1] return f"{preview}… ({len(value)} characters)" diff --git a/src/coretrace_python/reporters/json_format.py b/src/coretrace_python/reporters/json_format.py index ad5b94c..be04e3c 100644 --- a/src/coretrace_python/reporters/json_format.py +++ b/src/coretrace_python/reporters/json_format.py @@ -33,4 +33,16 @@ def render_json(report: Report) -> str: "tool": {"name": report.tool_name, "version": report.tool_version}, "findings": [finding_record(finding) for finding in report.findings], } + if report.coverage is not None: + coverage = report.coverage + document["coverage"] = { + "files": coverage.files, + "files_analysed": coverage.files_analysed, + "functions": coverage.functions, + "functions_analysed": coverage.functions_analysed, + "details": [ + {"path": d.path, "status": d.status, "functions": d.functions, "analysed": d.analysed} + for d in coverage.details + ], + } return json.dumps(document, indent=2) + "\n" diff --git a/src/coretrace_python/reporters/report.py b/src/coretrace_python/reporters/report.py index 593083a..90fbf42 100644 --- a/src/coretrace_python/reporters/report.py +++ b/src/coretrace_python/reporters/report.py @@ -4,7 +4,7 @@ from dataclasses import dataclass -from coretrace_python.findings import Finding +from coretrace_python.findings import Coverage, Finding def _order(finding: Finding) -> tuple[str, int, int, str]: @@ -17,6 +17,7 @@ class Report: findings: tuple[Finding, ...] tool_name: str tool_version: str + coverage: Coverage | None = None def __post_init__(self) -> None: object.__setattr__(self, "findings", tuple(sorted(self.findings, key=_order))) diff --git a/src/coretrace_python/reporters/text.py b/src/coretrace_python/reporters/text.py index 88003b4..06ba6c6 100644 --- a/src/coretrace_python/reporters/text.py +++ b/src/coretrace_python/reporters/text.py @@ -19,4 +19,6 @@ def render_text(report: Report) -> str: count = len(report.findings) summary = "no findings" if count == 0 else f"{count} finding{'s' if count > 1 else ''}" lines.append(summary) + if report.coverage is not None: + lines.append(report.coverage.summary()) return "\n".join(lines) + "\n" diff --git a/src/coretrace_python/source/__init__.py b/src/coretrace_python/source/__init__.py index bdba88f..28e8acd 100644 --- a/src/coretrace_python/source/__init__.py +++ b/src/coretrace_python/source/__init__.py @@ -1,7 +1,13 @@ """Source storage and location primitives.""" -from coretrace_python.source.manager import SourceManager +from coretrace_python.source.manager import SourceManager, decode_text from coretrace_python.source.model import SourceFile, SourceId, SourceSpan -__all__ = ["SourceFile", "SourceId", "SourceManager", "SourceSpan"] +__all__ = [ + "SourceFile", + "SourceId", + "SourceManager", + "SourceSpan", + "decode_text", +] diff --git a/src/coretrace_python/source/manager.py b/src/coretrace_python/source/manager.py index 038b9e9..1441bd0 100644 --- a/src/coretrace_python/source/manager.py +++ b/src/coretrace_python/source/manager.py @@ -20,6 +20,16 @@ def module_name_for(path: Path) -> str: return ".".join(reversed(parts)) or path.stem +def decode_text(data: bytes) -> str: + """Decode a text file, honouring a UTF-8 or UTF-16 byte order mark; Windows tools + such as ``pip freeze`` write UTF-16 with one.""" + + for mark, encoding in ((b"\xef\xbb\xbf", "utf-8-sig"), (b"\xff\xfe", "utf-16"), (b"\xfe\xff", "utf-16")): + if data.startswith(mark): + return data.decode(encoding) + return data.decode("utf-8") + + class SourceManager: """Own source text and return one canonical object for each source ID.""" @@ -49,8 +59,7 @@ def load_file(self, path: str | Path) -> SourceFile: if existing is not None: return existing - # utf-8-sig accepts ordinary UTF-8 and strips a leading UTF-8 BOM. - text = resolved_path.read_text(encoding="utf-8-sig") + text = decode_text(resolved_path.read_bytes()) source = SourceFile( source_id=source_id, text=text,