railiance-platform/tests/test_backup_credentials.py
codex 5ef016be01
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s
Contain backup upload credentials and prepare provider recovery gates
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
2026-09-05 19:21:06 +02:00

123 lines
4.7 KiB
Python

import os
from pathlib import Path
import subprocess
import unittest
ROOT = Path(__file__).resolve().parents[1]
class BackupCredentialTests(unittest.TestCase):
def run_shell(self, body):
env = {k: v for k, v in os.environ.items() if not k.startswith("RAILIANCE_BACKUP_")}
return subprocess.run(
["bash", "-c", 'source lib/railiance-backup-common.sh\n' + body],
cwd=ROOT, env=env, text=True, capture_output=True,
)
def test_missing_credentials_fail_with_safe_error(self):
result = self.run_shell(
'bao() { return 1; }\nrailiance_backup_require_openbao_lane'
)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, "")
self.assertIn("governed backup credential unavailable", result.stderr)
self.assertNotIn("unbound variable", result.stderr)
def test_explicit_governed_credentials_need_no_bao(self):
result = self.run_shell('''
bao() { echo unexpected-bao >&2; return 1; }
RAILIANCE_BACKUP_NC_TOKEN=test-only-placeholder
RAILIANCE_BACKUP_NC_WEBDAV_URL=https://example.invalid/upload
railiance_backup_require_openbao_lane
''')
self.assertEqual(result.returncode, 0)
self.assertEqual(result.stdout + result.stderr, "")
def test_curl_receives_credentials_only_on_stdin(self):
result = self.run_shell('''
RAILIANCE_BACKUP_NC_TOKEN=fixture-secret-marker
RAILIANCE_BACKUP_NC_WEBDAV_URL=https://example.invalid/fixture-secret-marker
curl() {
[[ "$*" != *fixture-secret-marker* ]] || return 9
local config
config="$(cat)"
[[ "$config" == *'user = "fixture-secret-marker:"'* ]] || return 8
[[ "$config" == *'url = "https://example.invalid/fixture-secret-marker/forgejo/test.age"'* ]] || return 7
printf 201
}
railiance_backup_nc_upload /tmp/test.age test.age
''')
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.stdout + result.stderr, '')
def test_redirect_and_backend_diagnostics_are_not_success_or_output(self):
for status in ['302', '401', '403', '500']:
result = self.run_shell('''
RAILIANCE_BACKUP_NC_TOKEN=fixture-secret-marker
RAILIANCE_BACKUP_NC_WEBDAV_URL=https://example.invalid/fixture-secret-marker
curl() { cat >/dev/null; echo fixture-secret-marker >&2; printf '%s'; }
railiance_backup_nc_upload /tmp/test.age test.age
''' % status)
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, '')
self.assertEqual(result.stderr, 'ERROR: Nextcloud upload failed\n')
def test_config_line_injection_fails_before_curl(self):
result = self.run_shell('''
RAILIANCE_BACKUP_NC_TOKEN=$'fixture\\noutput=/tmp/unwanted'
RAILIANCE_BACKUP_NC_WEBDAV_URL=https://example.invalid/upload
curl() { echo backend-was-called; return 1; }
railiance_backup_nc_upload /tmp/test.age test.age
''')
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, '')
self.assertEqual(result.stderr, 'ERROR: invalid backup upload input\n')
def test_plain_http_is_rejected_before_backend(self):
result = self.run_shell('''
RAILIANCE_BACKUP_NC_TOKEN=fixture
RAILIANCE_BACKUP_NC_WEBDAV_URL=http://example.invalid/upload
curl() { echo backend-was-called; return 1; }
railiance_backup_nc_upload /tmp/test.age test.age
''')
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, '')
def test_python_backend_exception_cannot_disclose_credential_url(self):
result = self.run_shell('''
RAILIANCE_BACKUP_NC_TOKEN=fixture-secret-marker
RAILIANCE_BACKUP_NC_WEBDAV_URL=https://example.invalid/fixture-secret-marker
command() {
if [[ "$*" == '-v curl' ]]; then return 1; fi
builtin command "$@"
}
python3() {
/usr/bin/python3 -c 'import sys,urllib.request
class FailedOpener:
def open(self, req, **kwargs):
raise RuntimeError(req.full_url)
urllib.request.build_opener=lambda *args: FailedOpener()
exec(sys.stdin.read())'
}
railiance_backup_nc_upload /dev/null test.age
''')
self.assertNotEqual(result.returncode, 0)
self.assertEqual(result.stdout, '')
self.assertEqual(result.stderr, 'ERROR: Nextcloud upload failed\n')
def test_openbao_lane_supplies_missing_credentials(self):
result = self.run_shell('''
bao() {
case "$*" in
"kv metadata get "*) return 0 ;;
"kv get -field=NC_WEBDAV_TOKEN "*) echo test-only-placeholder ;;
"kv get -field=NC_WEBDAV_URL "*) echo https://example.invalid/upload ;;
*) return 1 ;;
esac
}
railiance_backup_require_openbao_lane
[[ "$RAILIANCE_BACKUP_NC_TOKEN" == test-only-placeholder ]]
[[ "$RAILIANCE_BACKUP_NC_WEBDAV_URL" == https://example.invalid/upload ]]
''')
self.assertEqual(result.returncode, 0)
self.assertEqual(result.stdout + result.stderr, "")