189 lines
7.3 KiB
Python
189 lines
7.3 KiB
Python
"""Offline behavioral checks for deployment safety scripts (stdlib only)."""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parent
|
|
|
|
|
|
class ReleaseChecks(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.bin = self.root / 'bin'
|
|
self.bin.mkdir()
|
|
self.env = dict(os.environ, PATH=f'{self.bin}:{os.environ["PATH"]}',
|
|
REGISTRY='registry.example.invalid/test', GITHUB_REF='refs/heads/main',
|
|
GITHUB_OUTPUT=str(self.root / 'output'), REQUESTED_SHA='')
|
|
self.git('init', '-q', '-b', 'main')
|
|
self.git('config', 'user.email', 'ci@example.invalid')
|
|
self.git('config', 'user.name', 'CI Test')
|
|
self.commit('initial')
|
|
self.git('checkout', '-q', '-b', 'preprod')
|
|
self.commit('release')
|
|
self.preprod = self.git('rev-parse', 'HEAD')
|
|
self.git('checkout', '-q', 'main')
|
|
self.git('merge', '-q', '--no-ff', 'preprod', '-m', 'Promote')
|
|
self.env['GITHUB_SHA'] = self.git('rev-parse', 'HEAD')
|
|
self.env['VALIDATED_SHA'] = self.preprod
|
|
self.mock('docker', """for arg in "$@"; do
|
|
case "$arg" in
|
|
*:validated-preprod-$VALIDATED_SHA) printf 'Digest: sha256:%064d\\n' 0; exit 0 ;;
|
|
esac
|
|
done
|
|
exit 1
|
|
""")
|
|
|
|
def git(self, *args):
|
|
return subprocess.check_output(['git', *args], cwd=self.root, stderr=subprocess.PIPE,
|
|
text=True).strip()
|
|
|
|
def commit(self, text):
|
|
(self.root / 'source').write_text(text)
|
|
self.git('add', 'source')
|
|
self.git('commit', '-q', '-m', text)
|
|
|
|
def mock(self, name, body):
|
|
path = self.bin / name
|
|
path.write_text('#!/bin/sh\n' + body)
|
|
path.chmod(0o755)
|
|
|
|
def resolve(self):
|
|
return subprocess.run(['bash', str(SCRIPTS / 'resolve-release.sh')], cwd=self.root,
|
|
env=self.env, capture_output=True, text=True)
|
|
|
|
def test_normal_merge_promotes_second_parent(self):
|
|
result = self.resolve()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn(f'commit={self.preprod}', (self.root / 'output').read_text())
|
|
|
|
def test_same_sha_promotion(self):
|
|
self.env['GITHUB_SHA'] = self.preprod
|
|
self.assertEqual(self.resolve().returncode, 0)
|
|
|
|
def test_explicit_short_sha(self):
|
|
self.env['REQUESTED_SHA'] = self.preprod[:7]
|
|
self.assertEqual(self.resolve().returncode, 0)
|
|
|
|
def test_modified_production_tree_blocks_release(self):
|
|
self.commit('untested change')
|
|
self.env['GITHUB_SHA'] = self.git('rev-parse', 'HEAD')
|
|
self.env['REQUESTED_SHA'] = self.preprod
|
|
self.assertNotEqual(self.resolve().returncode, 0)
|
|
|
|
def test_missing_preprod_success_blocks_release(self):
|
|
self.env['VALIDATED_SHA'] = '0' * 40
|
|
self.assertNotEqual(self.resolve().returncode, 0)
|
|
|
|
def test_registry_failure_blocks_release(self):
|
|
self.mock('docker', 'exit 2\n')
|
|
self.assertNotEqual(self.resolve().returncode, 0)
|
|
|
|
def test_missing_exporter_marker_blocks_release(self):
|
|
self.mock('docker', """case "$*" in
|
|
*xpeditis-log-exporter*) exit 1 ;;
|
|
*) printf 'Digest: sha256:%064d\\n' 0 ;;
|
|
esac
|
|
""")
|
|
self.assertNotEqual(self.resolve().returncode, 0)
|
|
|
|
def test_invalid_digest_blocks_release(self):
|
|
self.mock('docker', "printf 'Digest: invalid\\n'\n")
|
|
self.assertNotEqual(self.resolve().returncode, 0)
|
|
|
|
def test_release_exports_pinned_digests(self):
|
|
self.assertEqual(self.resolve().returncode, 0)
|
|
outputs = (self.root / 'output').read_text()
|
|
self.assertIn('backend_digest=sha256:' + '0' * 64, outputs)
|
|
self.assertIn('log_exporter_digest=sha256:' + '0' * 64, outputs)
|
|
|
|
def test_input_is_never_executed(self):
|
|
self.env['REQUESTED_SHA'] = '$(touch injected)'
|
|
self.assertNotEqual(self.resolve().returncode, 0)
|
|
self.assertFalse((self.root / 'injected').exists())
|
|
|
|
def test_other_branch_cannot_deploy(self):
|
|
self.env['GITHUB_REF'] = 'refs/heads/dev'
|
|
self.assertNotEqual(self.resolve().returncode, 0)
|
|
|
|
def health(self, response, url='https://example.invalid/health'):
|
|
self.mock('curl', response)
|
|
self.mock('sleep', 'exit 0\n')
|
|
return subprocess.run(['bash', str(SCRIPTS / 'health-check.sh'), url],
|
|
env=self.env, capture_output=True, text=True)
|
|
|
|
def test_healthy_service(self):
|
|
self.assertEqual(self.health('printf 200\n').returncode, 0)
|
|
|
|
def test_unhealthy_service_blocks(self):
|
|
result = self.health('printf 503\n')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn('attempt 12', result.stdout)
|
|
|
|
def test_network_failure_blocks(self):
|
|
self.assertNotEqual(self.health('exit 7\n').returncode, 0)
|
|
|
|
def test_missing_https_blocks(self):
|
|
self.assertNotEqual(self.health('printf 200\n', '/api/v1/health').returncode, 0)
|
|
|
|
|
|
class AuditChecks(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.bin = self.root / 'bin'
|
|
self.bin.mkdir()
|
|
for project in ['backend', 'frontend', 'log-exporter']:
|
|
(self.root / 'apps' / project).mkdir(parents=True)
|
|
self.env = dict(os.environ, PATH=f'{self.bin}:{os.environ["PATH"]}',
|
|
RUNNER_TEMP=str(self.root), AUDIT_EXIT='0', SCAN_EXIT='0')
|
|
self.mock('timeout', 'shift\nexec "$@"\n')
|
|
self.mock('npm', """printf '%s' '{\"metadata\":{}}'
|
|
exit \"$AUDIT_EXIT\"
|
|
""")
|
|
self.mock('git', 'tar -cf - -T /dev/null\n')
|
|
self.mock('trivy', """while [ "$#" -gt 0 ]; do
|
|
if [ "$1" = --output ]; then shift; output="$1"; fi
|
|
shift
|
|
done
|
|
printf '%s' '{"Results":[{"Secrets":[{"RuleID":"fixture","Match":"synthetic-value","Code":{"Lines":[]}}]}]}' > "$output"
|
|
exit "$SCAN_EXIT"
|
|
""")
|
|
|
|
def mock(self, name, body):
|
|
path = self.bin / name
|
|
path.write_text('#!/bin/sh\n' + body)
|
|
path.chmod(0o755)
|
|
|
|
def audit(self):
|
|
return subprocess.run(['bash', str(SCRIPTS / 'security-audit.sh')],
|
|
cwd=self.root, env=self.env, capture_output=True, text=True)
|
|
|
|
def test_success_and_secret_redaction(self):
|
|
self.assertEqual(self.audit().returncode, 0)
|
|
report = (self.root / 'security-reports' / 'source-security.json').read_text()
|
|
self.assertNotIn('synthetic-value', report)
|
|
self.assertNotIn('Code', report)
|
|
self.assertIn('fixture', report)
|
|
|
|
def test_dependency_failure_is_not_masked_by_successful_source_scan(self):
|
|
self.env['AUDIT_EXIT'] = '1'
|
|
self.assertNotEqual(self.audit().returncode, 0)
|
|
self.assertEqual(len(list((self.root / 'security-reports').glob('npm-audit-*.json'))), 4)
|
|
|
|
def test_source_failure_is_not_masked_by_successful_dependency_audits(self):
|
|
self.env['SCAN_EXIT'] = '1'
|
|
self.assertNotEqual(self.audit().returncode, 0)
|
|
|
|
def test_timeout_fails_closed(self):
|
|
self.env['AUDIT_EXIT'] = '124'
|
|
self.assertNotEqual(self.audit().returncode, 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|