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_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] == 'stdin':
            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