repo-manager/tests/test_secret_scan.py

"""Real scanner integration on disposable Git indexes and never-valid credentials."""
import hashlib
import json
import os
import shlex
import subprocess
import sys
from pathlib import Path

import pytest

from repo_manager import secret_scan


@pytest.fixture
def binary():
    configured = os.environ.get('RMGR_TEST_GITLEAKS')
    if not configured:
        pytest.skip('set RMGR_TEST_GITLEAKS to the reviewed v8.30.1 Linux x64 binary')
    path = Path(configured)
    assert hashlib.sha256(path.read_bytes()).hexdigest() == secret_scan.BINARY_SHA256
    return path


@pytest.fixture
def repo(tmp_path):
    repo = tmp_path / 'repo'; repo.mkdir()
    git(repo, 'init', '-q')
    hooks = tmp_path / 'hooks'; hooks.mkdir()
    git(repo, 'config', 'core.hooksPath', str(hooks))
    git(repo, 'config', 'user.name', 'Disposable Test')
    git(repo, 'config', 'user.email', 'test@example.test')
    return repo


def git(repo, *args):
    return subprocess.check_output(['git', '-C', str(repo), *args], stderr=subprocess.DEVNULL)


@pytest.fixture
def private_key(tmp_path):
    path = tmp_path / 'disposable-key'
    subprocess.run(['ssh-keygen', '-q', '-t', 'ed25519', '-N', '', '-f', str(path)],
                   check=True, capture_output=True)
    return path.read_bytes()


def stage(repo, data, name='application.txt'):
    (repo / name).write_bytes(data)
    git(repo, 'add', '--', name)


def test_filename_digest_is_allowed_and_bare_assignment_stays_blocked(binary, repo):
    digest = os.urandom(32).hex()
    stage(repo, f"  cards/access.md: {digest}\n".encode(), "release.yaml")
    assert secret_scan.scan_staged(repo, binary)["status"] == "clean"
    git(repo, "rm", "-q", "--cached", "release.yaml")
    stage(repo, f"access: {digest}\n".encode(), "assignment.txt")
    assert secret_scan.scan_staged(repo, binary)["status"] == "blocked"


def test_clean_staged_blob(binary, repo):
    stage(repo, b'print("hello")\n')
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'clean'
    assert result['scanned_blobs'] == 1


def test_real_secret_commit_is_refused_without_disclosure(binary, repo, private_key):
    hook = Path(git(repo, 'rev-parse', '--git-path', 'hooks/pre-commit').decode().strip())
    script = str(Path(secret_scan.__file__))
    hook.write_text('#!/bin/sh\nexec ' + ' '.join(shlex.quote(s) for s in
                    [sys.executable, script, '--path', str(repo), '--binary', str(binary)]) + '\n')
    hook.chmod(0o700)
    stage(repo, private_key)
    result = subprocess.run(['git', 'commit', '-qm', 'synthetic detection probe'],
                            cwd=repo, capture_output=True, check=False)
    assert result.returncode != 0
    assert bool(private_key.strip() in result.stdout + result.stderr) is False
    assert b'"status": "blocked"' in result.stdout + result.stderr
    assert subprocess.run(['git', 'rev-parse', '--verify', 'HEAD'], cwd=repo,
                          capture_output=True, check=False).returncode != 0


def test_staged_secret_refused_even_if_worktree_is_clean(binary, repo, private_key):
    stage(repo, private_key)
    (repo / 'application.txt').write_text('clean worktree\n')
    assert secret_scan.scan_staged(repo, binary)['status'] == 'blocked'


def test_unstaged_secret_not_misrepresented_as_index_scan(binary, repo, private_key):
    stage(repo, b'clean staged data\n')
    (repo / 'application.txt').write_bytes(private_key)
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'clean'
    assert result['scope'] == 'changed-staged-blobs'


def test_repo_config_environment_and_inline_allow_cannot_disable_scan(binary, repo, private_key, monkeypatch):
    (repo / '.gitleaks.toml').write_text('[allowlist]\nregexes=[".*"]\n')
    (repo / '.gitleaksignore').write_text('*\n')
    monkeypatch.setenv('GITLEAKS_CONFIG', str(repo / '.gitleaks.toml'))
    monkeypatch.setenv('GITLEAKS_CONFIG_TOML', '[allowlist]\nregexes=[".*"]\n')
    stage(repo, b'# gitleaks:allow\n' + private_key)
    assert secret_scan.scan_staged(repo, binary)['status'] == 'blocked'


def test_symlink_content_is_not_dereferenced(binary, repo, private_key, tmp_path):
    target = tmp_path / 'outside-secret'; target.write_bytes(private_key)
    (repo / 'link').symlink_to(target)
    git(repo, 'add', '--', 'link')
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'clean'
    assert result['scanned_blobs'] == 1


def test_missing_scanner_is_unverified(repo):
    stage(repo, b'clean\n')
    result = secret_scan.scan_staged(repo, repo / 'missing')
    assert result['status'] == 'unverified'
    assert secret_scan.exit_code(result) == 2


def test_wrong_digest_is_refused_before_execution(repo):
    marker = repo / 'executed'
    fake = repo / 'fake'; fake.write_text('#!/bin/sh\ntouch ' + str(marker) + '\n'); fake.chmod(0o700)
    result = secret_scan.scan_staged(repo, fake)
    assert result['reason'] == 'scanner-digest-mismatch'
    assert not marker.exists()


def test_changed_submodule_is_unverified(binary, repo):
    stage(repo, b'clean\n'); git(repo, 'commit', '-qm', 'seed')
    commit = git(repo, 'rev-parse', 'HEAD').decode().strip()
    git(repo, 'update-index', '--add', '--cacheinfo', '160000,' + commit + ',submodule')
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'unsupported-index-entry'


def test_oversize_blob_refuses_instead_of_skipping(binary, repo, monkeypatch):
    monkeypatch.setattr(secret_scan, 'MAX_BLOB_BYTES', 8)
    stage(repo, b'0123456789')
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'staged-byte-limit'


def test_git_replace_cannot_hide_staged_secret(binary, repo, private_key):
    stage(repo, private_key)
    secret = git(repo, 'rev-parse', ':application.txt').decode().strip()
    clean = subprocess.check_output(['git', 'hash-object', '-w', '--stdin'], input=b'clean\n', cwd=repo).decode().strip()
    git(repo, 'replace', secret, clean)
    assert secret_scan.scan_staged(repo, binary)['status'] == 'blocked'


def test_index_change_during_scan_refuses(binary, repo, monkeypatch):
    stage(repo, b'clean\n')
    original = secret_scan.git
    count = 0
    def changing(repo, *args):
        nonlocal count
        if args == ('ls-files', '--stage', '-z'):
            count += 1
            if count == 2: stage(repo, b'changed after scan\n')
        return original(repo, *args)
    monkeypatch.setattr(secret_scan, 'git', changing)
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'index-changed-during-scan'


def test_explicit_git_index_file_is_honored(binary, repo, private_key, monkeypatch, tmp_path):
    stage(repo, b'clean default index\n')
    default = Path(git(repo, 'rev-parse', '--git-path', 'index').decode().strip())
    if not default.is_absolute(): default = repo / default
    other = tmp_path / 'alternate-index'; other.write_bytes(default.read_bytes())
    monkeypatch.setenv('GIT_INDEX_FILE', str(other))
    stage(repo, private_key)
    assert secret_scan.scan_staged(repo, binary)['status'] == 'blocked'
    monkeypatch.delenv('GIT_INDEX_FILE')
    assert secret_scan.scan_staged(repo, binary)['status'] == 'clean'


def test_cli_outputs_only_summary(binary, repo, private_key):
    stage(repo, private_key)
    result = subprocess.run([sys.executable, '-m', 'repo_manager.secret_scan', '--path', str(repo),
                             '--binary', str(binary)], capture_output=True, check=False)
    assert result.returncode == 1
    assert json.loads(result.stdout)['status'] == 'blocked'
    assert result.stderr == b''
    assert bool(private_key.strip() in result.stdout) is False


@pytest.mark.parametrize('failure', ['exit', 'timeout'])
def test_scanner_failure_never_passes_or_emits_diagnostics(binary, repo, monkeypatch, failure):
    stage(repo, b'clean\n')
    original = secret_scan.subprocess.run
    def broken(command, **kwargs):
        if len(command) > 1 and command[1] in ('stdin', 'dir'):
            if failure == 'timeout':
                raise subprocess.TimeoutExpired(command, 20, output=b'synthetic-sensitive-diagnostic')
            return subprocess.CompletedProcess(command, 2)
        return original(command, **kwargs)
    monkeypatch.setattr(secret_scan.subprocess, 'run', broken)
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'unverified'
    assert 'synthetic-sensitive-diagnostic' not in json.dumps(result)


def test_actual_commit_only_hook_scans_its_temporary_index(binary, repo, private_key):
    stage(repo, b'clean\n'); stage(repo, b'also clean\n', 'unrelated.txt')
    git(repo, 'commit', '-qm', 'clean baseline')
    hook = Path(git(repo, 'rev-parse', '--git-path', 'hooks/pre-commit').decode().strip())
    hook.write_text('#!/bin/sh\nexec ' + ' '.join(shlex.quote(s) for s in
                    [sys.executable, str(Path(secret_scan.__file__)), '--path', str(repo),
                     '--binary', str(binary)]) + '\n')
    hook.chmod(0o700)
    stage(repo, b'unrelated staged change\n', 'unrelated.txt')
    (repo / 'application.txt').write_bytes(private_key)
    before = git(repo, 'rev-parse', 'HEAD')
    result = subprocess.run(['git', 'commit', '--only', '-qm', 'synthetic selective commit', '--',
                             'application.txt'], cwd=repo, capture_output=True, check=False)
    assert result.returncode != 0
    assert git(repo, 'rev-parse', 'HEAD') == before
    assert bool(private_key.strip() in result.stdout + result.stderr) is False
    assert b'"status": "blocked"' in result.stdout + result.stderr


def test_empty_and_deletion_only_index(binary, repo):
    assert secret_scan.scan_staged(repo, binary)['scanned_blobs'] == 0
    stage(repo, b'clean\n'); git(repo, 'commit', '-qm', 'seed')
    assert secret_scan.scan_staged(repo, binary)['scanned_blobs'] == 0
    git(repo, 'rm', '-q', 'application.txt')
    assert secret_scan.scan_staged(repo, binary)['scanned_blobs'] == 0


def test_duplicate_blobs_and_exact_attribution(binary, repo, private_key):
    stage(repo, private_key, 'secret.png')  # original filename cannot allowlist content
    stage(repo, private_key, 'duplicate.txt')
    stage(repo, b'clean\n', 'clean.txt')
    oid = git(repo, 'rev-parse', ':secret.png').decode().strip()
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'blocked'
    assert result['scanned_blobs'] == 2
    assert result['blocked_blob_ids'] == [oid]


@pytest.mark.parametrize('size', [4096, 4097, 100000, 100001])
def test_small_batch_and_large_fallback_equivalence(binary, repo, private_key, size):
    data = b'\n' * (size - len(private_key)) + private_key
    stage(repo, data)
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'blocked'
    assert result['blocked_blob_ids'] == [git(repo, 'rev-parse', ':application.txt').decode().strip()]


def test_file_boundaries_never_synthesize_a_finding(binary, repo, private_key):
    cut = private_key.index(b'\n') // 2
    stage(repo, private_key[:cut], 'first')
    stage(repo, private_key[cut:], 'second')
    # Each independent fragment matches the original single-blob stdin outcome;
    # concatenation would synthesize the complete private-key opening marker.
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'clean'
    assert result['scanned_blobs'] == 2


def test_binary_and_encoded_fixture_equivalence(binary, repo, private_key, monkeypatch):
    import base64
    original = secret_scan.DIRECTORY_BLOB_BYTES
    for data in [b'\x00\x01\x02\n' + private_key, base64.b64encode(private_key)]:
        stage(repo, data)
        monkeypatch.setattr(secret_scan, 'DIRECTORY_BLOB_BYTES', 0)
        stdin = secret_scan.scan_staged(repo, binary)
        monkeypatch.setattr(secret_scan, 'DIRECTORY_BLOB_BYTES', original)
        batched = secret_scan.scan_staged(repo, binary)
        assert batched == stdin
        assert batched['status'] == 'blocked'


def test_batch_count_bound_and_independent_findings(binary, repo, private_key, monkeypatch):
    monkeypatch.setattr(secret_scan, 'MAX_BATCH_BLOBS', 3)
    for i in range(7):
        stage(repo, (f'fixture {i}\n'.encode() + private_key) if i % 2 else f'clean {i}\n'.encode(), str(i))
    result = secret_scan.scan_staged(repo, binary)
    expected = sorted(git(repo, 'rev-parse', ':' + str(i)).decode().strip() for i in [1, 3, 5])
    assert result['status'] == 'blocked'
    assert result['scanned_blobs'] == 7
    assert result['blocked_blob_ids'] == expected


def test_total_and_object_count_bounds_refuse(binary, repo, monkeypatch):
    stage(repo, b'12345', 'a'); stage(repo, b'67890', 'b')
    monkeypatch.setattr(secret_scan, 'MAX_TOTAL_BYTES', 9)
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'staged-byte-limit'
    monkeypatch.setattr(secret_scan, 'MAX_BLOBS', 1)
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'staged-blob-limit'


@pytest.mark.parametrize('mutation', ['path', 'in-place'])
def test_verified_executable_snapshot_survives_mutation(binary, repo, monkeypatch, tmp_path, mutation):
    import shutil
    scanner = tmp_path / 'scanner'; shutil.copyfile(binary, scanner); scanner.chmod(0o700)
    stage(repo, b'clean\n')
    original = secret_scan.staged_objects
    def mutate(repo):
        if mutation == 'path':
            scanner.unlink()
        scanner.write_bytes(b'not executable scanner bytes\n')
        scanner.chmod(0o700)
        return original(repo)
    monkeypatch.setattr(secret_scan, 'staged_objects', mutate)
    assert secret_scan.scan_staged(repo, scanner)['status'] == 'clean'


@pytest.mark.parametrize('fault', ['warning', 'unknown-path', 'missing-report', 'returncode'])
def test_partial_scans_and_invalid_reports_refuse(binary, repo, monkeypatch, fault):
    stage(repo, b'clean\n')
    original = secret_scan.subprocess.run
    def fail(command, **kwargs):
        result = original(command, **kwargs)
        if command[1] == 'dir':
            report = Path(command[command.index('--report-path') + 1])
            if fault == 'warning':
                kwargs['stderr'].write(b'synthetic-sensitive-diagnostic\n')
            elif fault == 'unknown-path':
                report.write_text('blobs/unknown\n')
            elif fault == 'missing-report':
                report.unlink()
            else:
                return subprocess.CompletedProcess(command, 20)
        return result
    monkeypatch.setattr(secret_scan.subprocess, 'run', fail)
    result = secret_scan.scan_staged(repo, binary)
    assert result['status'] == 'unverified'
    assert 'synthetic-sensitive-diagnostic' not in json.dumps(result)


def test_deadline_exhaustion_refuses(binary, repo, monkeypatch):
    stage(repo, b'clean\n')
    monkeypatch.setattr(secret_scan, 'SCAN_SECONDS', 0)
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'scan-deadline-exceeded'


def test_git_object_output_corruption_refuses(binary, repo, monkeypatch):
    stage(repo, b'clean\n')
    original = secret_scan.subprocess.run
    def corrupt(command, **kwargs):
        result = original(command, **kwargs)
        if command[-1] == '--batch':
            output = kwargs['stdout']
            output.seek(-3, 2); output.write(b'xx\n')
        return result
    monkeypatch.setattr(secret_scan.subprocess, 'run', corrupt)
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'blob-digest-mismatch'


def test_report_growth_limit_refuses(binary, repo, private_key, monkeypatch):
    stage(repo, private_key)
    monkeypatch.setattr(secret_scan, 'MAX_REPORT_BYTES', 16)
    assert secret_scan.scan_staged(repo, binary)['status'] == 'unverified'


def test_sealed_executable_rejects_write(binary, repo, monkeypatch):
    import errno
    stage(repo, b'clean\n')
    original = secret_scan.subprocess.run
    checked = []
    def assert_sealed(command, **kwargs):
        if command[1] in ('dir', 'stdin'):
            fd = kwargs['pass_fds'][0]
            with pytest.raises(OSError) as error:
                os.pwrite(fd, b'X', 0)
            assert error.value.errno == errno.EPERM
            checked.append(True)
        return original(command, **kwargs)
    monkeypatch.setattr(secret_scan.subprocess, 'run', assert_sealed)
    assert secret_scan.scan_staged(repo, binary)['status'] == 'clean'
    assert checked


def test_nonvolatile_storage_refuses(binary, repo, monkeypatch):
    stage(repo, b'clean\n')
    monkeypatch.setattr(secret_scan, 'is_tmpfs', lambda directory: False)
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'volatile-storage-unavailable'


def test_cleanup_watchdog_loss_refuses(binary, repo, monkeypatch):
    stage(repo, b'clean\n')
    original = secret_scan.object_batch
    def kill_watchdog(repo, objects, directory):
        # Discover only our child watchdog using the context manager wrapper.
        watcher.kill(); watcher.wait(timeout=5)
        return original(repo, objects, directory)
    original_storage = secret_scan.private_storage
    import contextlib
    @contextlib.contextmanager
    def storage():
        nonlocal watcher
        with original_storage() as pair:
            watcher = pair[1]
            yield pair
    watcher = None
    monkeypatch.setattr(secret_scan, 'private_storage', storage)
    monkeypatch.setattr(secret_scan, 'object_batch', kill_watchdog)
    assert secret_scan.scan_staged(repo, binary)['reason'] == 'cleanup-watchdog-unavailable'


def test_parent_sigkill_triggers_watchdog_cleanup(tmp_path):
    import time
    script = ('from repo_manager.secret_scan import private_storage; import time\n'
              'with private_storage() as (directory, watcher):\n'
              ' (directory / "fixture").write_bytes(b"public generated watchdog fixture")\n'
              ' print(str(directory), flush=True)\n'
              ' time.sleep(30)\n')
    process = subprocess.Popen([sys.executable, '-c', script], stdout=subprocess.PIPE,
                               stderr=subprocess.DEVNULL, text=True)
    try:
        import select
        ready, _, _ = select.select([process.stdout], [], [], 5)
        assert ready
        directory = Path(process.stdout.readline().strip())
        assert directory.is_dir()
        process.kill(); process.wait(timeout=5)
        until = time.monotonic() + 5
        while directory.exists() and time.monotonic() < until:
            time.sleep(.02)
        assert not directory.exists()
    finally:
        if process.poll() is None:
            process.kill(); process.wait(timeout=5)
        process.stdout.close()


def test_watchdog_deadline_quiesces_exact_parent_before_cleanup():
    import select
    import time
    script = ('from repo_manager import secret_scan as s; import time\n'
              's._CLEANUP = s._CLEANUP.replace("[], [], 180)", "[], [], .2)")\n'
              'with s.private_storage() as (directory, watcher):\n'
              ' (directory / "fixture").write_bytes(b"public generated fixture")\n'
              ' print(str(directory), flush=True)\n'
              ' time.sleep(30)\n')
    process = subprocess.Popen([sys.executable, '-c', script], stdout=subprocess.PIPE,
                               stderr=subprocess.DEVNULL, text=True)
    try:
        ready, _, _ = select.select([process.stdout], [], [], 5)
        assert ready
        directory = Path(process.stdout.readline().strip())
        assert directory.is_dir()
        assert process.wait(timeout=5) < 0
        until = time.monotonic() + 5
        while directory.exists() and time.monotonic() < until:
            time.sleep(.02)
        assert not directory.exists()
    finally:
        if process.poll() is None:
            process.kill(); process.wait(timeout=5)
        process.stdout.close()


def test_parent_death_kills_source_child():
    import select
    script = ('from repo_manager.secret_scan import child_limits\n'
              'import functools, os, subprocess, sys, time\n'
              'child = subprocess.Popen([sys.executable, "-c", "import time;time.sleep(30)"], '
              'preexec_fn=functools.partial(child_limits, 1024, os.getpid()))\n'
              'print(child.pid, flush=True)\n'
              'time.sleep(30)\n')
    process = subprocess.Popen([sys.executable, '-c', script], stdout=subprocess.PIPE,
                               stderr=subprocess.DEVNULL, text=True)
    descriptor = None
    try:
        ready, _, _ = select.select([process.stdout], [], [], 5)
        assert ready
        child_pid = int(process.stdout.readline())
        descriptor = os.pidfd_open(child_pid)
        process.kill(); process.wait(timeout=5)
        dead, _, _ = select.select([descriptor], [], [], 5)
        assert dead
    finally:
        if descriptor is not None:
            os.close(descriptor)
        if process.poll() is None:
            process.kill(); process.wait(timeout=5)
        process.stdout.close()