tegwick
006dc23fdf
isolated-authelia.py runs restored LLDAP as the network-none anchor with Authelia joining its namespace. Checks the original config, storage key, issuer key, KeyCape's client secret and existing-user first factor, with wrong-password and wrong-client-secret refusal. Confinement rewrites: LDAP to loopback, file notifier, NTP check off, healthcheck file off. Imports isolated-lldap.py unedited; the T38-pinned KeyCape sources are untouched. Native fixture tests on the pinned images pass (9). T17 now awaits the founder-attended run. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 328925@bnt-lap001 Assistant-Session: 14b795fd-c632-4c01-8d28-e7e490c7b0ed
242 lines
13 KiB
Python
242 lines
13 KiB
Python
"""Isolated Authelia restoration with synthetic retained credentials only."""
|
|
import base64
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import subprocess
|
|
import uuid
|
|
|
|
import pytest
|
|
import yaml
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from test_identity_bundle import stage # noqa: F401 (fixture reuse)
|
|
from test_isolated_lldap_recovery import PASSWORD, retained_directory # noqa: F401 (fixture reuse)
|
|
import test_isolated_lldap_recovery as lldap_fixture
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location('isolated_authelia', ROOT / 'tools/bootstrap-recovery/isolated-authelia.py')
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
MANIFESTS = ROOT / 'sso-mfa/k8s/authelia'
|
|
CLIENT_SECRET = 'synthetic-keycape-client-' + secrets.token_hex(16) # Disposable fixture only.
|
|
|
|
|
|
def secret_values():
|
|
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
return {'jwt_secret': secrets.token_hex(32), 'session_secret': secrets.token_hex(32),
|
|
'storage_encryption_key': secrets.token_hex(32), 'ldap_password': PASSWORD,
|
|
'oidc_hmac_secret': secrets.token_hex(32),
|
|
'oidc_issuer_private_key': key.private_bytes(serialization.Encoding.PEM,
|
|
serialization.PrivateFormat.TraditionalOpenSSL,
|
|
serialization.NoEncryption()).decode(),
|
|
'keycape_client_secret_hash': 'placeholder-replaced-in-native-fixture'}
|
|
|
|
|
|
def resources(root, values):
|
|
"""Write the repository's own Authelia manifests as archived resources."""
|
|
def write(kind, name, value):
|
|
p = root / 'resources/sso' / kind
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
value = {**value, 'metadata': {**value.get('metadata', {}), 'name': name, 'namespace': 'sso'}}
|
|
(p / (name + '.json')).write_text(json.dumps(value))
|
|
deployment = next(d for d in yaml.safe_load_all((MANIFESTS / 'deployment.yaml').read_text())
|
|
if d and d['kind'] == 'Deployment')
|
|
configmap = yaml.safe_load((MANIFESTS / 'configmap.yaml').read_text())
|
|
config = yaml.safe_load(configmap['data']['configuration.yml'])
|
|
ldap = config['authentication_backend']['ldap']
|
|
ldap.update(base_dn='dc=example,dc=test', user='uid=admin,ou=people,dc=example,dc=test')
|
|
configmap['data']['configuration.yml'] = yaml.safe_dump(config, sort_keys=False)
|
|
write('deployment', 'authelia', deployment)
|
|
write('configmap', 'authelia-config', configmap)
|
|
write('secret', 'authelia-secrets', {'data': {k: base64.b64encode(v.encode()).decode() for k, v in values.items()}})
|
|
write('secret', 'authelia-smtp', {'data': {'password': base64.b64encode(b'synthetic-smtp-canary').decode()}})
|
|
keycape = {'authelia': {'clientId': 'keycape', 'clientSecret': CLIENT_SECRET,
|
|
'browserBaseURL': 'https://auth.coulomb.social'}}
|
|
write('secret', 'keycape-config', {'data': {'config.yaml': base64.b64encode(yaml.safe_dump(keycape).encode()).decode()}})
|
|
store = root / 'data/authelia/db.sqlite3'
|
|
store.parent.mkdir(parents=True, exist_ok=True)
|
|
if not store.exists():
|
|
store.write_bytes(b'')
|
|
|
|
|
|
def test_configuration_keeps_originals_and_confines_ldap_and_notifier(tmp_path):
|
|
values = secret_values()
|
|
resources(tmp_path, values)
|
|
runtime = m.configuration(tmp_path)
|
|
config = yaml.safe_load(runtime['config'])
|
|
assert config['authentication_backend']['ldap']['url'] == m.LDAP_URL
|
|
assert config['notifier'] == {'filesystem': {'filename': '/var/authelia/data/' + m.NOTIFICATIONS}}
|
|
assert config['ntp']['disable_startup_check'] is True
|
|
assert config['server']['disable_healthcheck'] is True
|
|
assert config['identity_providers']['oidc']['clients'][0]['secret'].startswith('{{ secret ')
|
|
assert runtime['secrets']['/run/secrets/authelia/storage_encryption_key'] == values['storage_encryption_key'].encode()
|
|
assert not any('smtp' in name.lower() for name in runtime['env'])
|
|
assert all(path.startswith('/run/secrets/authelia/') for name, path in runtime['env'].items()
|
|
if name.endswith('_FILE'))
|
|
assert runtime['host'] == 'auth.coulomb.social' and runtime['client_secret'] == CLIENT_SECRET
|
|
for value in values.values():
|
|
assert value.encode() not in runtime['config']
|
|
|
|
|
|
def test_configuration_refuses_missing_secret_file(tmp_path):
|
|
values = secret_values()
|
|
del values['storage_encryption_key']
|
|
resources(tmp_path, values)
|
|
with pytest.raises(m.bundle.Refused):
|
|
m.configuration(tmp_path)
|
|
|
|
|
|
def test_configuration_refuses_env_from_unknown_source(tmp_path):
|
|
resources(tmp_path, secret_values())
|
|
path = tmp_path / 'resources/sso/deployment/authelia.json'
|
|
value = json.loads(path.read_text())
|
|
value['spec']['template']['spec']['containers'][0]['env'].append(
|
|
{'name': 'AUTHELIA_EXTRA', 'valueFrom': {'secretKeyRef': {'name': 'other', 'key': 'x'}}})
|
|
path.write_text(json.dumps(value))
|
|
with pytest.raises(m.bundle.Refused):
|
|
m.configuration(tmp_path)
|
|
|
|
|
|
def test_configuration_refuses_unknown_keycape_client(tmp_path):
|
|
resources(tmp_path, secret_values())
|
|
path = tmp_path / 'resources/sso/secret/keycape-config.json'
|
|
keycape = {'authelia': {'clientId': 'not-registered', 'clientSecret': CLIENT_SECRET,
|
|
'browserBaseURL': 'https://auth.coulomb.social'}}
|
|
path.write_text(json.dumps({'metadata': {'name': 'keycape-config', 'namespace': 'sso'}, 'data': {'config.yaml': base64.b64encode(yaml.safe_dump(keycape).encode()).decode()}}))
|
|
with pytest.raises(m.bundle.Refused):
|
|
m.configuration(tmp_path)
|
|
|
|
|
|
def test_public_jwk_thumbprint_is_rfc7638():
|
|
pem = secret_values()['oidc_issuer_private_key'].encode()
|
|
jwk, thumbprint = m.public_jwk(pem)
|
|
assert set(jwk) == {'e', 'kty', 'n'} and len(thumbprint) == 43
|
|
|
|
|
|
@pytest.fixture
|
|
def retained_authelia(retained_directory): # noqa: F811
|
|
if os.environ.get('NK49_RUN_ISOLATED_AUTHELIA_TEST') != '1':
|
|
pytest.skip('explicit local disposable-image fixture opt-in')
|
|
m.preflight()
|
|
values = secret_values()
|
|
digest = m.command(['docker', 'run', '--rm', '--pull=never', '--network', 'none', '--entrypoint', '/app/authelia',
|
|
m.IMAGE, 'crypto', 'hash', 'generate', 'pbkdf2', '--variant', 'sha512',
|
|
'--password', CLIENT_SECRET]).decode()
|
|
values['keycape_client_secret_hash'] = digest.split('Digest: ', 1)[1].strip()
|
|
resources(retained_directory, values)
|
|
runtime = m.configuration(retained_directory)
|
|
(retained_directory / 'data/authelia/db.sqlite3').unlink()
|
|
name = 'nk49-authelia-fixture-' + uuid.uuid4().hex
|
|
try:
|
|
m.command(['docker', 'run', '--detach', '--pull=never', '--name', name, '--network', 'none',
|
|
'--read-only', '--cap-drop', 'ALL', '--user', '1000:1000',
|
|
*m.tmpfs('/var/authelia/data', 67108864), *m.tmpfs('/run/secrets/authelia', 1048576),
|
|
*m.tmpfs('/config', 1048576), *[x for k, v in runtime['env'].items() for x in ('--env', k + '=' + v)],
|
|
'--entrypoint', '/bin/sh', m.IMAGE, '-c', 'sleep 600'])
|
|
m.command(['docker', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/run/secrets/authelia'],
|
|
input=m.archive_files({k.rsplit('/', 1)[1]: v for k, v in runtime['secrets'].items()}))
|
|
m.command(['docker', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/config'],
|
|
input=m.archive_files({'configuration.yml': runtime['config']}))
|
|
m.command(['docker', 'exec', name, '/app/authelia', 'storage', 'migrate', 'up',
|
|
'--config', '/config/configuration.yml'])
|
|
store = m.command(['docker', 'exec', name, 'cat', '/var/authelia/data/db.sqlite3'])
|
|
(retained_directory / 'data/authelia/db.sqlite3').write_bytes(store)
|
|
yield retained_directory, values
|
|
finally:
|
|
subprocess.run(['docker', 'rm', '--force', '--volumes', name], capture_output=True, timeout=30)
|
|
|
|
|
|
def test_native_restored_authelia_passes_all_checks(retained_authelia):
|
|
root, values = retained_authelia
|
|
result = {}
|
|
m.restore_authelia(root, 'admin', PASSWORD, result)
|
|
assert result['status'] == 'verified', result
|
|
for key in ('original_configuration_validated', 'storage_encryption_key_verified', 'issuer_key_matches_archive',
|
|
'existing_user_first_factor', 'wrong_password_refused', 'archived_client_secret_accepted',
|
|
'wrong_client_secret_refused', 'only_loopback_interface_verified', 'container_cleanup_verified',
|
|
'container_isolation_verified_before_private_import', 'notifications_confined_to_tmpfs_file'):
|
|
assert result[key] is True, key
|
|
assert result['grant_issued'] is False
|
|
text = json.dumps(result)
|
|
assert PASSWORD not in text and CLIENT_SECRET not in text
|
|
assert not any(v in text for v in values.values())
|
|
|
|
|
|
def test_native_wrong_password_fails_and_removes_containers(retained_authelia):
|
|
root, _ = retained_authelia
|
|
result = {}
|
|
with pytest.raises(m.bundle.Refused):
|
|
m.restore_authelia(root, 'admin', 'wrong-retained-password', result)
|
|
assert result['failed_stage'] == 'existing_user_first_factor'
|
|
assert result['container_cleanup_verified']
|
|
|
|
|
|
def test_native_wrong_storage_key_is_refused(retained_authelia):
|
|
root, _ = retained_authelia
|
|
path = root / 'resources/sso/secret/authelia-secrets.json'
|
|
value = json.loads(path.read_text())
|
|
value['data']['storage_encryption_key'] = base64.b64encode(secrets.token_hex(32).encode()).decode()
|
|
path.write_text(json.dumps(value))
|
|
result = {}
|
|
with pytest.raises(m.bundle.Refused):
|
|
m.restore_authelia(root, 'admin', PASSWORD, result)
|
|
assert result['failed_stage'] == 'storage_decryption'
|
|
assert result['container_cleanup_verified']
|
|
|
|
|
|
assert lldap_fixture.m.IMAGE == m.lldap.IMAGE
|
|
|
|
|
|
def test_encrypted_bundle_to_native_authelia_and_tmpfs_cleanup(retained_authelia, stage, tmp_path, monkeypatch, capsys):
|
|
import shutil
|
|
import sys
|
|
root, values = retained_authelia
|
|
# Replace the synthetic placeholders with the retained LLDAP/Authelia state,
|
|
# keeping the complete profile's resource identities and other key assets.
|
|
for component in ('lldap', 'authelia'):
|
|
for source in (root / 'data' / component).iterdir():
|
|
if source.is_file():
|
|
shutil.copyfile(source, stage / 'data' / component / source.name)
|
|
for kind, name in (('deployment', 'lldap'), ('secret', 'lldap-secrets'), ('deployment', 'authelia'),
|
|
('configmap', 'authelia-config'), ('secret', 'authelia-secrets'), ('secret', 'authelia-smtp')):
|
|
target = stage / 'resources/sso' / kind / (name + '.json')
|
|
value = json.loads(target.read_text())
|
|
retained = json.loads((root / 'resources/sso' / kind / (name + '.json')).read_text())
|
|
value.update({k: v for k, v in retained.items() if k != 'metadata'})
|
|
target.write_text(json.dumps(value))
|
|
target = stage / 'resources/sso/secret/keycape-config.json'
|
|
value = json.loads(target.read_text())
|
|
value['data']['config.yaml'] = json.loads(
|
|
(root / 'resources/sso/secret/keycape-config.json').read_text())['data']['config.yaml']
|
|
target.write_text(json.dumps(value))
|
|
key = tmp_path / 'fixture-age.key'
|
|
m.command(['age-keygen', '-o', str(key)])
|
|
recipient = m.command(['age-keygen', '-y', str(key)]).decode().strip()
|
|
profile = m.bundle.profile()
|
|
profile['public_recipient'] = recipient
|
|
monkeypatch.setattr(m.bundle, 'profile', lambda: profile)
|
|
artifact = tmp_path / 'net-kingdom-identity-20261016T063159Z-0123456789ab.tar.age'
|
|
m.bundle.seal(stage, artifact, {'synthetic_native_fixture': True})
|
|
transfer = tmp_path / 'transfer.json'
|
|
transfer.write_text(json.dumps({'schema': 'netkingdom.identity-offsite.v1', 'status': 'readback_verified',
|
|
'provider': 'Storage Share', 'folder': 'Backup/railiance-backups/net-kingdom-identity',
|
|
'artifact': artifact.name, 'ciphertext_bytes': artifact.stat().st_size,
|
|
'ciphertext_sha256': m.offline.offsite.digest(artifact), 'etag': '"synthetic"',
|
|
'fetched_path': str(artifact), 'independent_owner_retrieval_verified': True}))
|
|
monkeypatch.setattr(m.verify.custody, 'private_key_fd', lambda: os.open(key, os.O_RDONLY))
|
|
monkeypatch.setattr(m.offline, 'private_prompt', lambda label: PASSWORD)
|
|
receipt = tmp_path / 'recovery.json'
|
|
monkeypatch.setattr(sys, 'argv', ['isolated-authelia', '--transfer-receipt', str(transfer),
|
|
'--receipt', str(receipt), '--user', 'admin'])
|
|
assert m.main() == 0, receipt.read_text()
|
|
result = json.loads(receipt.read_text())
|
|
assert result['status'] == 'isolated_authelia_verified'
|
|
assert result['temporary_plaintext_cleanup_verified'] and result['container_cleanup_verified']
|
|
assert result['all_manifest_hashes_verified'] and result['issuer_key_matches_archive']
|
|
assert not result['production_contacted'] and not result['secret_values_exported']
|
|
text = capsys.readouterr().out + receipt.read_text()
|
|
assert PASSWORD not in text and CLIENT_SECRET not in text
|
|
assert not any(v in text for v in values.values())
|