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
24 changes: 24 additions & 0 deletions .dagger/tests/test_module_fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,18 @@ def test_should_contract_tamper_failure_before_provider_transport() -> None:
assert all(all(marker in source for marker in markers) for source in sources)


def test_should_run_exact_alma_detector_adversary_in_python_fixture() -> None:
# Given the hosted generated-client fixture and its production guard
python = _python_source()
guard = _foundation_guard_source()

# When / Then the hosted contract runs every exact pinned Alma proof
assert "await _alma_gitleaks_contract()" in python
assert "76cfea53cb96d215278048a326bd4aab91af9949" in python
assert _gitleaks_image(python) == _gitleaks_image(guard)
assert all(marker in python for marker in ("fresh-copy", "snapshot-copy"))


def test_should_run_both_fixture_checks_from_the_explicit_root_source() -> None:
# Given an explicit root source and observable dynamic module checks
events: list[str] = []
Expand Down Expand Up @@ -226,6 +238,18 @@ def _python_source() -> str:
return (FIXTURES / "python_consumer/.dagger/src/python_consumer/main.py").read_text()


def _foundation_guard_source() -> str:
path = ROOT / "modules/portfolio-foundation/.dagger/src/portfolio_foundation/guard.py"
return path.read_text()


def _gitleaks_image(source: str) -> str:
pattern = r"ghcr\.io/gitleaks/gitleaks:v8\.30\.1@sha256:[0-9a-f]+"
match = re.search(pattern, source.replace('"\n "', ""))
assert match is not None
return match.group()


def _typescript_source() -> str:
return (FIXTURES / "typescript_consumer/src/index.ts").read_text()

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@
"ef8299f97635c4c30e2298f48f30763ab782a4ad2c95b744649439a039421e36"
)
GITLEAKS_IMAGE: Final = (
"ghcr.io/gitleaks/gitleaks:v8.29.1@sha256:"
"aa036a2f4bdfe3cc3c55fa4326308efabb4a6be498c883c864fd1d0d5585438a"
"ghcr.io/gitleaks/gitleaks:v8.30.1@sha256:"
"c00b6bd0aeb3071cbcb79009cb16a60dd9e0a7c60e2be9ab65d25e6bc8abbb7f"
)
ACTIONLINT_PATH: Final = "/usr/local/bin/actionlint"
CANARY_EXIT_CODE: Final = 86
Expand Down Expand Up @@ -76,7 +76,7 @@ def _snapshot_commands() -> tuple[str, ...]:
return (
'test -n "$(find /snapshot -type f -print -quit)"',
"echo guard-snapshot-nonempty >&2",
_gitleaks("/snapshot", no_git=True),
*_configured_gitleaks("/snapshot", no_git=True),
)


Expand All @@ -88,20 +88,31 @@ def _history_commands(commit_sha: str) -> tuple[str, ...]:
'test -n "$(git -C /repo rev-list --all)"',
"git -C /repo fsck --full --no-dangling",
"echo guard-history-verified >&2",
_gitleaks("/repo", log_options="--all"),
*_configured_gitleaks("/repo", log_options="--all"),
)


def _configured_gitleaks(
source: str, *, no_git: bool = False, log_options: str | None = None
) -> tuple[str, ...]:
config = f"{source}/.gitleaks.toml"
configured = _gitleaks(source, no_git=no_git, log_options=log_options, config=config)
default = _gitleaks(source, no_git=no_git, log_options=log_options)
return (f"if test -f {config}; then", f" {configured}", "else", f" {default}", "fi")


def _gitleaks(
source: str,
*,
no_git: bool = False,
exit_code: int | None = None,
log_options: str | None = None,
config: str | None = None,
) -> str:
options: tuple[str, ...] = ("gitleaks", "detect", "--source", source)
options += _no_git_flag(no_git)
options += _log_options_flag(log_options)
options += _config_flag(config)
options += ("--redact", "--no-banner")
options += _exit_code_flag(exit_code)
return " ".join(options)
Expand All @@ -119,6 +130,12 @@ def _log_options_flag(options: str | None) -> tuple[str, ...]:
return ()


def _config_flag(path: str | None) -> tuple[str, ...]:
if path is not None:
return ("--config", path)
return ()


def _exit_code_flag(exit_code: int | None) -> tuple[str, ...]:
if exit_code is not None:
return ("--exit-code", str(exit_code))
Expand Down
11 changes: 11 additions & 0 deletions modules/portfolio-foundation/.dagger/tests/test_guard.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,17 @@ def test_should_scan_nonempty_snapshot_and_all_canonical_history() -> None:
assert "--source /repo --log-opts=--all" in command


def test_should_apply_repository_config_to_snapshot_and_history() -> None:
# Given
command = secret_scan_command("c" * 40)

# When / Then
assert "test -f /snapshot/.gitleaks.toml" in command
assert "--config /snapshot/.gitleaks.toml" in command
assert "test -f /repo/.gitleaks.toml" in command
assert "--config /repo/.gitleaks.toml" in command


def test_should_pin_every_guard_tool_image_by_digest() -> None:
# Given
images = (ACTIONLINT_IMAGE, GITLEAKS_IMAGE)
Expand Down
168 changes: 168 additions & 0 deletions modules/portfolio-foundation/.dagger/tests/test_guard_integration.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,13 @@
from __future__ import annotations

import re
import shutil
import subprocess
from dataclasses import dataclass
from pathlib import Path

from portfolio_foundation.guard import GITLEAKS_IMAGE


@dataclass(frozen=True)
class GuardFixture:
Expand All @@ -20,8 +24,17 @@ class GuardFixture:
"kristof-mattei/km-crates-publish-test",
"711c3b50ce63192b88f22215793ca7f1eeb7b439",
)
ALMA_FIXTURES = GuardFixture("hseshadr/almamesh", "55dba578dde057d1dd6d680b5f978280498ef0ea")
MODULE = Path(__file__).parents[2]
DAGGER_GUARD = ("dagger", "--progress=logs", "-m", ".", "call", "guard")
FIXTURE_PATHS = (
Path("backend/tests/test_predictive_golden.py"),
Path("frontend/packages/browser/integration/parity.mjs"),
)
PAYLOAD_PARTS = (
"616c6d616d6573682d7061726974792d",
"666978747572652d7369676e65723030",
)


def _run(*args: str, cwd: Path) -> subprocess.CompletedProcess[str]:
Expand Down Expand Up @@ -60,6 +73,108 @@ def _output(result: subprocess.CompletedProcess[str]) -> str:
return result.stdout


def _write_alma_fixtures(root: Path) -> None:
seed = "".join(PAYLOAD_PARTS)
python, javascript = (root / path for path in FIXTURE_PATHS)
python.parent.mkdir(parents=True, exist_ok=True)
javascript.parent.mkdir(parents=True, exist_ok=True)
python.write_text(f'FIXTURE_KEY_SEED_HEX = "{seed}"\n')
javascript.write_text(f'const PREDICTIVE_FIXTURE_KEY_SEED_HEX =\n "{seed}";\n')


def _git(root: Path, *arguments: str) -> str:
result = _run("git", *arguments, cwd=root)
_require(result, "git " + " ".join(arguments))
return result.stdout.strip()


def _commit(root: Path, message: str) -> str:
_git(root, "add", ".")
_git(root, "commit", "-qm", message)
return _git(root, "rev-parse", "HEAD")


def _configure_author(root: Path) -> None:
_git(root, "config", "user.email", "fixture@example.invalid")
_git(root, "config", "user.name", "Fixture")


def _policy_prefix(baseline: Path | None) -> str:
if baseline is None:
return 'title = "AlmaMesh fixture policy"\n[extend]\nuseDefault = true\n'
return baseline.read_text().replace("\n[allowlist]\n", "\n[[allowlists]]\n")


def _write_allowlist(root: Path, commit_sha: str, baseline: Path | None = None) -> None:
lines = (
"[[allowlists]]",
'targetRules = ["generic-api-key"]',
'condition = "AND"',
f'commits = ["{commit_sha}"]',
"paths = [",
" '''^backend/tests/test_predictive_golden\\.py$''',",
" '''^frontend/packages/browser/integration/parity\\.mjs$''',",
"]",
)
policy = _policy_prefix(baseline).rstrip() + "\n\n" + "\n".join(lines) + "\n"
(root / ".gitleaks.toml").write_text(policy)


def _alma_policy(root: Path, repository: Path) -> Path:
policy = root / "policy"
policy.mkdir()
_write_allowlist(policy, ALMA_FIXTURES.commit_sha, repository / ".gitleaks.toml")
return policy / ".gitleaks.toml"


def _fresh_copy_repository(root: Path) -> Path:
repository = _materialize(root, ALMA_FIXTURES)
_configure_author(repository)
for path in FIXTURE_PATHS:
(repository / path).unlink()
_commit(repository, "remove historical fixtures")
_write_alma_fixtures(repository)
_commit(repository, "fresh fixture copies")
return repository


def _detector_arguments(history: bool, config: Path | None) -> tuple[str, ...]:
mode = ("--log-opts=--all",) if history else ("--no-git",)
options = ("--config", "/policy/.gitleaks.toml") if config else ()
return (
"detect",
"--source",
"/repo",
*mode,
*options,
"--redact",
"--no-banner",
"--verbose",
)


def _detect(
root: Path, *, history: bool, config: Path | None = None
) -> subprocess.CompletedProcess[str]:
docker = shutil.which("docker")
assert docker is not None, "Docker is required for the real detector contract"
command = (docker, "run", "--rm", "-v", f"{root}:/repo:ro", GITLEAKS_IMAGE)
config_mount = ("-v", f"{config}:/policy/.gitleaks.toml:ro") if config else ()
return subprocess.run( # noqa: S603
(*command[:-1], *config_mount, command[-1], *_detector_arguments(history, config)),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
check=False,
)


def _assert_two_generic_api_keys(result: subprocess.CompletedProcess[str]) -> None:
assert result.returncode != 0, result.stdout
assert "leaks found: 2" in result.stdout
assert len(re.findall(r"RuleID:\s+generic-api-key", result.stdout)) == 2


def _assert_success_evidence(result: subprocess.CompletedProcess[str]) -> None:
output = _output(result)
assert result.returncode == 0, output
Expand Down Expand Up @@ -118,3 +233,56 @@ def test_should_reject_secret_retained_only_in_history(tmp_path: Path) -> None:

# Then
_assert_history_secret(result)


def test_should_detect_both_alma_fixture_patterns_without_allowlist(tmp_path: Path) -> None:
# Given
snapshot = tmp_path / "snapshot"
snapshot.mkdir()
_write_alma_fixtures(snapshot)

# When
result = _detect(snapshot, history=False)

# Then
_assert_two_generic_api_keys(result)


def test_should_allow_only_the_exact_historical_alma_matches(tmp_path: Path) -> None:
# Given
repository = _materialize(tmp_path, ALMA_FIXTURES)
policy = _alma_policy(tmp_path, repository)

# When
result = _detect(repository, history=True, config=policy)

# Then
assert result.returncode == 0, result.stdout
assert "no leaks found" in result.stdout


def test_should_reject_a_fresh_commit_copy_despite_historical_allowlist(tmp_path: Path) -> None:
# Given
repository = _fresh_copy_repository(tmp_path)
policy = _alma_policy(tmp_path, repository)

# When
result = _detect(repository, history=True, config=policy)

# Then
_assert_two_generic_api_keys(result)


def test_should_reject_a_snapshot_copy_despite_historical_allowlist(tmp_path: Path) -> None:
# Given
repository = _materialize(tmp_path, ALMA_FIXTURES)
snapshot = tmp_path / "snapshot"
snapshot.mkdir()
_write_alma_fixtures(snapshot)
policy = _alma_policy(tmp_path, repository)

# When
result = _detect(snapshot, history=False, config=policy)

# Then
_assert_two_generic_api_keys(result)
Loading