Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
38 changes: 38 additions & 0 deletions plugins/secrets/config_secrets/config_secrets.py
Original file line number Diff line number Diff line change
@@ -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
9 changes: 9 additions & 0 deletions plugins/secrets/config_secrets/plugin.toml
Original file line number Diff line number Diff line change
@@ -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"
42 changes: 10 additions & 32 deletions plugins/secrets/hardcoded_secrets/hardcoded_secrets.py
Original file line number Diff line number Diff line change
@@ -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
6 changes: 4 additions & 2 deletions src/coretrace_python/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
56 changes: 39 additions & 17 deletions src/coretrace_python/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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:
Expand All @@ -218,24 +241,14 @@ 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
graph = graph.merge(parse_dependencies(sources.add_source(str(path), text)))
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] = (),
Expand Down Expand Up @@ -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

Expand All @@ -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(
Expand Down Expand Up @@ -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))
Expand All @@ -390,6 +411,7 @@ def analyze_project(
MappingProxyType(keys),
reused,
advisories,
Coverage(tuple(sorted(coverage, key=lambda c: c.path))),
)


Expand Down Expand Up @@ -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)
10 changes: 9 additions & 1 deletion src/coretrace_python/findings/__init__.py
Original file line number Diff line number Diff line change
@@ -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",
]
45 changes: 45 additions & 0 deletions src/coretrace_python/findings/coverage.py
Original file line number Diff line number Diff line change
@@ -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"
)
2 changes: 2 additions & 0 deletions src/coretrace_python/interprocedural/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from coretrace_python.interprocedural.modulegraph import (
ModuleGraph,
build_module_graph,
discover_files,
discover_sources,
project_symbol,
)
Expand Down Expand Up @@ -42,6 +43,7 @@
"Target",
"UnknownTarget",
"build_module_graph",
"discover_files",
"discover_sources",
"project_symbol",
]
Loading
Loading