|
| 1 | +"""Executable admission and lane contracts for the fixed pg-erd consumer.""" |
| 2 | + |
| 3 | +import re |
| 4 | +import subprocess |
| 5 | +from pathlib import Path |
| 6 | + |
| 7 | +import pytest |
| 8 | +import yaml |
| 9 | + |
| 10 | +ROOT = Path(__file__).resolve().parents[1] |
| 11 | +PATH = ROOT / '.github/workflows/pg-erd-cloud-quality.yml' |
| 12 | + |
| 13 | + |
| 14 | +def workflow(): |
| 15 | + """Load with BaseLoader so GitHub's on key is not parsed as YAML 1.1 bool.""" |
| 16 | + return yaml.load(PATH.read_text(), Loader=yaml.BaseLoader) |
| 17 | + |
| 18 | + |
| 19 | +def test_fixed_reusable_scope(): |
| 20 | + """No arbitrary executable inputs, secrets, or cancellation of its caller.""" |
| 21 | + data = workflow() |
| 22 | + assert data['on'] == {'workflow_call': ''} |
| 23 | + assert data['permissions'] == {'contents': 'read'} |
| 24 | + assert 'concurrency' not in data |
| 25 | + assert set(data['jobs']) == {'backend', 'frontend'} |
| 26 | + for job in data['jobs'].values(): |
| 27 | + assert job['runs-on'] == {'group': 'CWL CI isolated', 'labels': ['self-hosted', 'linux', 'x64', 'cwlab-ci-isolated']} |
| 28 | + assert job['timeout-minutes'] == '30' |
| 29 | + assert 'if' not in job |
| 30 | + assert job['steps'][0]['name'] == 'Admit fixed caller before checkout' |
| 31 | + assert 'uses' not in job['steps'][0] |
| 32 | + assert job['steps'][1]['uses'].startswith('step-security/harden-runner@') |
| 33 | + checkout = job['steps'][2] |
| 34 | + assert checkout['with']['persist-credentials'] == 'false' |
| 35 | + assert checkout['with']['ref'] == '${{ github.event.pull_request.head.sha || github.sha }}' |
| 36 | + for step in job['steps']: |
| 37 | + assert not step.get('continue-on-error') |
| 38 | + if 'uses' in step: |
| 39 | + assert re.search(r'@[0-9a-f]{40}$', step['uses']) |
| 40 | + |
| 41 | + |
| 42 | +@pytest.mark.parametrize('job_name', ['backend', 'frontend']) |
| 43 | +@pytest.mark.parametrize('event,repository,head,base,ref,expected', [ |
| 44 | + ('pull_request', 'ContextualWisdomLab/pg-erd-cloud', 'ContextualWisdomLab/pg-erd-cloud', 'ContextualWisdomLab/pg-erd-cloud', 'refs/pull/1/merge', 0), |
| 45 | + ('pull_request', 'ContextualWisdomLab/pg-erd-cloud', 'attacker/fork', 'ContextualWisdomLab/pg-erd-cloud', 'refs/pull/1/merge', 1), |
| 46 | + ('push', 'ContextualWisdomLab/pg-erd-cloud', '', '', 'refs/heads/main', 0), |
| 47 | + ('push', 'ContextualWisdomLab/pg-erd-cloud', '', '', 'refs/heads/feature', 1), |
| 48 | + ('workflow_dispatch', 'ContextualWisdomLab/pg-erd-cloud', '', '', 'refs/heads/main', 0), |
| 49 | + ('workflow_dispatch', 'ContextualWisdomLab/pg-erd-cloud', '', '', 'refs/heads/feature', 1), |
| 50 | + ('pull_request_target', 'ContextualWisdomLab/pg-erd-cloud', 'ContextualWisdomLab/pg-erd-cloud', 'ContextualWisdomLab/pg-erd-cloud', 'refs/heads/main', 1), |
| 51 | + ('push', 'ContextualWisdomLab/.github', '', '', 'refs/heads/main', 1), |
| 52 | + ('pull_request', 'ContextualWisdomLab/pg-erd-cloud', 'ContextualWisdomLab/pg-erd-cloud', 'attacker/base', 'refs/pull/1/merge', 1), |
| 53 | +]) |
| 54 | +def test_actual_admission_shell(job_name, event, repository, head, base, ref, expected): |
| 55 | + """Run the actual first step with allowed and rejected caller tuples.""" |
| 56 | + step = workflow()['jobs'][job_name]['steps'][0] |
| 57 | + assert step['env'] == { |
| 58 | + 'CALLER_REPOSITORY': '${{ github.repository }}', 'CALLER_EVENT': '${{ github.event_name }}', |
| 59 | + 'PR_HEAD_REPOSITORY': '${{ github.event.pull_request.head.repo.full_name }}', |
| 60 | + 'PR_BASE_REPOSITORY': '${{ github.event.pull_request.base.repo.full_name }}', |
| 61 | + 'CALLER_REF': '${{ github.ref }}', |
| 62 | + } |
| 63 | + env = {'PATH': '/usr/bin:/bin', 'CALLER_REPOSITORY': repository, 'CALLER_EVENT': event, |
| 64 | + 'PR_HEAD_REPOSITORY': head, 'PR_BASE_REPOSITORY': base, 'CALLER_REF': ref} |
| 65 | + result = subprocess.run(['/bin/bash', '--noprofile', '--norc', '-e', '-o', 'pipefail', '-c', step['run']], env=env, capture_output=True, text=True, timeout=5) |
| 66 | + assert result.returncode == expected |
| 67 | + assert 'attacker' not in result.stdout + result.stderr |
| 68 | + |
| 69 | + |
| 70 | +def test_backend_environment_is_job_private(): |
| 71 | + """Candidate dependencies cannot be installed into a shared runner Python.""" |
| 72 | + steps = workflow()['jobs']['backend']['steps'] |
| 73 | + install = next(s for s in steps if 'pip install' in s.get('run', '')) |
| 74 | + script = install['run'] |
| 75 | + assert 'mktemp -d' in script |
| 76 | + assert 'python -m venv' in script |
| 77 | + assert 'trap ' in script and ' EXIT' in script |
| 78 | + assert 'PIP_NO_CACHE_DIR' in script |
| 79 | + assert install['shell'] == 'bash --noprofile --norc -e -o pipefail {0}' |
| 80 | + |
| 81 | + |
| 82 | +@pytest.mark.parametrize('failed_tool, expected', [('', 0), ('mypy', 17), ('pytest', 17)]) |
| 83 | +def test_backend_shell_cleanup_and_failure_propagation(tmp_path, failed_tool, expected): |
| 84 | + """Execute the actual lane with synthetic tools; never install dependencies.""" |
| 85 | + checkout = tmp_path / 'checkout'; checkout.mkdir() |
| 86 | + (checkout / 'backend').mkdir() |
| 87 | + runner_temp = tmp_path / 'runner-temp'; runner_temp.mkdir() |
| 88 | + sentinel = runner_temp / 'unrelated'; sentinel.write_text('keep') |
| 89 | + tools = tmp_path / 'tools'; tools.mkdir() |
| 90 | + trace = tmp_path / 'trace' |
| 91 | + python_stub = tools / 'python' |
| 92 | + python_stub.write_text('''#!/bin/bash |
| 93 | +set -eu |
| 94 | +printf 'python %s\\n' "$*" >> "$TRACE" |
| 95 | +if [ "${1-}" = '-m' ] && [ "${2-}" = 'venv' ]; then |
| 96 | + mkdir -p "$3/bin" |
| 97 | + ln -s "$STUB_PYTHON" "$3/bin/python" |
| 98 | +fi |
| 99 | +''') |
| 100 | + python_stub.chmod(0o755) |
| 101 | + for tool in ['actionlint', 'shellcheck', 'mypy', 'pytest']: |
| 102 | + path = tools / tool |
| 103 | + path.write_text(f'''#!/bin/bash |
| 104 | +printf '{tool}\\n' >> "$TRACE" |
| 105 | +if [ "$FAILED_TOOL" = '{tool}' ]; then exit 17; fi |
| 106 | +''') |
| 107 | + path.chmod(0o755) |
| 108 | + step = next(s for s in workflow()['jobs']['backend']['steps'] if 'pip install' in s.get('run', '')) |
| 109 | + env = {'PATH': f'{tools}:/usr/bin:/bin', 'RUNNER_TEMP': str(runner_temp), |
| 110 | + 'TRACE': str(trace), 'STUB_PYTHON': str(python_stub), 'FAILED_TOOL': failed_tool} |
| 111 | + result = subprocess.run(['/bin/bash', '--noprofile', '--norc', '-e', '-o', 'pipefail', '-c', step['run']], cwd=checkout, env=env, capture_output=True, text=True, timeout=5) |
| 112 | + assert result.returncode == expected, result.stderr |
| 113 | + assert sorted(p.name for p in runner_temp.iterdir()) == ['unrelated'] |
| 114 | + assert sentinel.read_text() == 'keep' |
| 115 | + lines = trace.read_text().splitlines() |
| 116 | + assert sum('pip install --require-hashes' in line for line in lines) == 2 |
| 117 | + assert 'python -m unittest discover -s .github/tests -v' in lines |
| 118 | + if failed_tool == 'mypy': |
| 119 | + assert 'pytest' not in lines |
| 120 | + else: |
| 121 | + assert 'pytest' in lines |
| 122 | + |
| 123 | + |
| 124 | +def test_preserved_product_commands(): |
| 125 | + """Centralization keeps every existing backend/frontend validation command.""" |
| 126 | + data = workflow() |
| 127 | + backend = data['jobs']['backend']; frontend = data['jobs']['frontend'] |
| 128 | + commands = '\n'.join(s.get('run', '') for s in backend['steps']) |
| 129 | + for command in ['python -m pip install --require-hashes -r requirements-dev.lock', |
| 130 | + 'python -m pip install --require-hashes -r ../.github/requirements.lock', |
| 131 | + 'python -m unittest discover -s .github/tests -v', |
| 132 | + 'python .github/scripts/validate_codeql_backfill.py', |
| 133 | + 'actionlint .github/workflows/*.yml', 'shellcheck .github/scripts/*.sh', |
| 134 | + 'mypy app', 'pytest -q']: |
| 135 | + assert command in commands |
| 136 | + frontend_commands = '\n'.join(s.get('run', '') for s in frontend['steps']) |
| 137 | + for command in ['npm ci --include=dev', 'npm run typecheck', 'npm run test', 'npm run build']: |
| 138 | + assert command in frontend_commands |
| 139 | + assert frontend['env']['NODE_ENV'] == 'test' |
| 140 | + assert next(s for s in frontend['steps'] if s['name'] == 'Build')['env']['NODE_ENV'] == 'production' |
| 141 | + assert all('runner.' not in str(v) for j in data['jobs'].values() for v in j.get('env', {}).values()) |
0 commit comments