Prepare native isolated directory recovery and offline service images
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
This commit is contained in:
parent
31f483966c
commit
d5484bb90a
9 changed files with 815 additions and 7 deletions
177
tests/test_isolated_lldap_recovery.py
Normal file
177
tests/test_isolated_lldap_recovery.py
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
"""Real isolated-image restoration with synthetic retained credentials only."""
|
||||
import base64
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tarfile
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from test_identity_bundle import stage
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
spec = importlib.util.spec_from_file_location('isolated', ROOT / 'tools/bootstrap-recovery/isolated-lldap.py')
|
||||
m = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(m)
|
||||
PASSWORD = 'synthetic-retained-password'
|
||||
JWT = secrets.token_hex(32) # Disposable fixture only; never provider custody.
|
||||
|
||||
|
||||
def resources(root):
|
||||
def write(kind, value):
|
||||
p = root / 'resources/sso' / kind
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
(p / ('lldap-secrets.json' if kind == 'secret' else 'lldap.json')).write_text(json.dumps(value))
|
||||
write('secret', {'metadata': {'name': 'lldap-secrets', 'namespace': 'sso'}, 'data': {
|
||||
'LLDAP_JWT_SECRET': base64.b64encode(JWT.encode()).decode(),
|
||||
'LLDAP_LDAP_USER_PASS': base64.b64encode(PASSWORD.encode()).decode()}})
|
||||
write('deployment', {'metadata': {'name': 'lldap', 'namespace': 'sso'}, 'spec': {'template': {'spec': {
|
||||
'containers': [{'name': 'lldap', 'env': [{'name': 'LLDAP_LDAP_BASE_DN', 'value': 'dc=example,dc=test'}]}]}}}})
|
||||
|
||||
|
||||
def test_configuration_never_generates_a_missing_original_key(tmp_path):
|
||||
resources(tmp_path)
|
||||
(tmp_path / 'data/lldap').mkdir(parents=True)
|
||||
with pytest.raises(m.bundle.Refused):
|
||||
m.configuration(tmp_path)
|
||||
|
||||
|
||||
def test_configuration_preserves_original_seed_and_disables_reset_sending(tmp_path):
|
||||
resources(tmp_path)
|
||||
data = tmp_path / 'data/lldap'
|
||||
data.mkdir(parents=True)
|
||||
(data / 'lldap_config.toml').write_text('key_seed = "synthetic-original-seed"\n')
|
||||
import tomllib
|
||||
value, jwt, password, base = m.configuration(tmp_path)
|
||||
config = tomllib.loads(value.decode())
|
||||
assert config['key_seed'] == 'synthetic-original-seed'
|
||||
assert not config['force_ldap_user_pass_reset']
|
||||
assert not config['force_update_private_key']
|
||||
assert not config['smtp_options']['enable_password_reset']
|
||||
assert PASSWORD.encode() not in value and jwt not in value
|
||||
|
||||
|
||||
def test_tar_transport_refuses_path_escape():
|
||||
with pytest.raises(m.bundle.Refused):
|
||||
m.archive_files({'../escape': b'synthetic'})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def retained_directory(tmp_path):
|
||||
if os.environ.get('NK49_RUN_ISOLATED_LLDAP_TEST') != '1':
|
||||
pytest.skip('explicit local disposable-image fixture opt-in')
|
||||
m.preflight()
|
||||
resources(tmp_path)
|
||||
name = 'nk49-lldap-fixture-' + uuid.uuid4().hex
|
||||
config = ('database_url = "sqlite:///data/users.db?mode=rwc"\n'
|
||||
'ldap_base_dn = "dc=example,dc=test"\n'
|
||||
'ldap_host = "127.0.0.1"\nhttp_host = "127.0.0.1"\n'
|
||||
'key_seed = "synthetic-retained-key-seed"\n'
|
||||
'jwt_secret = "' + JWT + '"\n'
|
||||
'ldap_user_pass = "' + PASSWORD + '"\n')
|
||||
try:
|
||||
m.command(['docker', 'run', '--detach', '--pull=never', '--name', name,
|
||||
'--network', 'none', '--read-only', '--cap-drop', 'ALL', '--user', '1000:1000',
|
||||
'--tmpfs', '/data:rw,nosuid,nodev,size=67108864,uid=1000,gid=1000',
|
||||
'--tmpfs', '/tmp:rw,nosuid,nodev,size=16777216,uid=1000,gid=1000',
|
||||
'--entrypoint', '/bin/sh', m.IMAGE, '-c', 'sleep 1800'])
|
||||
m.command(['docker', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/data'],
|
||||
input=m.archive_files({'lldap_config.toml': config.encode()}))
|
||||
m.command(['docker', 'exec', '--detach', name, '/bin/sh', '-c',
|
||||
'echo $$ > /tmp/server.pid; exec /app/lldap run --config-file /data/lldap_config.toml > /dev/null 2>&1'])
|
||||
result = m.command(['docker', 'run', '--interactive', '--rm', '--pull=never', '--network', 'container:' + name,
|
||||
'--read-only', '--entrypoint', 'python3', m.CLIENT, '-c', m.PROBE],
|
||||
input=json.dumps({'user': 'admin', 'password': PASSWORD,
|
||||
'base': 'dc=example,dc=test', 'wrong_password': 'invalid-fixture'}).encode())
|
||||
assert json.loads(result)['status'] == 'verified', result.decode()
|
||||
m.command(['docker', 'exec', name, '/bin/sh', '-c', 'kill -TERM "$(cat /tmp/server.pid)"'])
|
||||
for _ in range(60):
|
||||
check = subprocess.run(['docker', 'exec', name, '/bin/sh', '-c', 'kill -0 "$(cat /tmp/server.pid)"'],
|
||||
capture_output=True, timeout=10)
|
||||
if check.returncode:
|
||||
break
|
||||
time.sleep(.1)
|
||||
else:
|
||||
pytest.fail('fixture did not stop')
|
||||
raw = m.command(['docker', 'exec', name, 'tar', '-cf', '-', '-C', '/data', '.'])
|
||||
data = tmp_path / 'data/lldap'
|
||||
data.mkdir(parents=True)
|
||||
with tarfile.open(fileobj=io.BytesIO(raw)) as archive:
|
||||
for entry in archive:
|
||||
if entry.isdir():
|
||||
continue
|
||||
assert entry.isfile() and not entry.name.startswith('/') and '..' not in Path(entry.name).parts
|
||||
target = data / entry.name.removeprefix('./')
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_bytes(archive.extractfile(entry).read())
|
||||
yield tmp_path
|
||||
finally:
|
||||
subprocess.run(['docker', 'rm', '--force', '--volumes', name], capture_output=True, timeout=30)
|
||||
|
||||
|
||||
def test_native_restored_directory_binds_existing_user_and_preserves_rows(retained_directory):
|
||||
result = {}
|
||||
m.restore_lldap(retained_directory, 'admin', PASSWORD, result)
|
||||
assert result['status'] == 'verified'
|
||||
assert result['existing_user_native_ldap_bind'] and result['wrong_password_refused']
|
||||
assert result['directory_identity_rows_preserved'] and result['container_cleanup_verified']
|
||||
assert result['only_loopback_interface_verified'] and result['external_routes_absent']
|
||||
assert PASSWORD not in json.dumps(result)
|
||||
|
||||
|
||||
def test_native_wrong_retained_password_fails_and_removes_containers(retained_directory):
|
||||
result = {}
|
||||
with pytest.raises(m.bundle.Refused):
|
||||
m.restore_lldap(retained_directory, 'admin', 'wrong-retained-password', result)
|
||||
assert result['container_cleanup_verified']
|
||||
assert result['failed_stage'] == 'existing_user_bind'
|
||||
assert not result.get('existing_user_native_ldap_bind')
|
||||
|
||||
|
||||
def test_encrypted_bundle_to_native_directory_and_tmpfs_cleanup(retained_directory, stage, tmp_path, monkeypatch, capsys):
|
||||
import shutil
|
||||
import sys
|
||||
# Replace the synthetic SQLite placeholder with a real retained directory,
|
||||
# retaining the complete profile's resource identities and other key assets.
|
||||
for source in (retained_directory / 'data/lldap').iterdir():
|
||||
if source.is_file():
|
||||
shutil.copyfile(source, stage / 'data/lldap' / source.name)
|
||||
dep = stage / 'resources/sso/deployment/lldap.json'
|
||||
value = json.loads(dep.read_text())
|
||||
value['spec'] = json.loads((retained_directory / 'resources/sso/deployment/lldap.json').read_text())['spec']
|
||||
dep.write_text(json.dumps(value))
|
||||
secret = stage / 'resources/sso/secret/lldap-secrets.json'
|
||||
value = json.loads(secret.read_text())
|
||||
value['data'] = json.loads((retained_directory / 'resources/sso/secret/lldap-secrets.json').read_text())['data']
|
||||
secret.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-lldap', '--transfer-receipt', str(transfer),
|
||||
'--receipt', str(receipt), '--user', 'admin'])
|
||||
assert m.main() == 0
|
||||
result = json.loads(receipt.read_text())
|
||||
assert result['status'] == 'isolated_lldap_verified'
|
||||
assert result['temporary_plaintext_cleanup_verified'] and result['container_cleanup_verified']
|
||||
assert result['all_manifest_hashes_verified'] and result['directory_identity_rows_preserved']
|
||||
assert not result['production_contacted'] and not result['secret_values_exported']
|
||||
assert PASSWORD not in capsys.readouterr().out + receipt.read_text()
|
||||
Loading…
Add table
Add a link
Reference in a new issue