net-kingdom/tests/test_identity_offline_recovery.py

"""Outage download boundaries, failure cleanup and independently usable kit."""
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import subprocess
import sys
import urllib.error
import zipfile

import pytest

ROOT = Path(__file__).resolve().parents[1]


def load(name, root=ROOT):
    spec = importlib.util.spec_from_file_location(name, root / 'tools/bootstrap-recovery' / (name + '.py'))
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


m = load('retrieve-identity-offline')
k = load('recovery-kit')


class Response(io.BytesIO):
    def __init__(self, url, raw, etag='"original"', status=200):
        super().__init__(raw)
        self.url = url
        self.status = status
        self.headers = {'Content-Length': str(len(raw)), 'ETag': etag}


class Provider:
    def __init__(self, body):
        self.body = body
        self.calls = []
        self.tamper = False
        self.etag = '"original"'
        self.objects = {}

    def open(self, request, timeout):
        method = request.get_method()
        self.calls.append((method, request.full_url, dict(request.headers)))
        assert request.full_url.startswith((m.ACCOUNT_BASE, m.offsite.BASE))
        if self.objects or method == 'PUT':
            if method == 'PUT':
                assert request.get_header('If-none-match') == '*'
                assert request.full_url not in self.objects
                self.objects[request.full_url] = request.data
                return Response(request.full_url, b'', status=201)
            if request.full_url not in self.objects:
                raise urllib.error.HTTPError(request.full_url, 404, 'missing', {}, None)
            return Response(request.full_url, self.objects[request.full_url])
        raw = self.body if not self.tamper or method == 'HEAD' else b'X' * len(self.body)
        return Response(request.full_url, raw, self.etag)


@pytest.fixture
def record():
    return dict(schema='netkingdom.identity-offsite.v1', status='readback_verified',
                provider='Storage Share', folder='Backup/railiance-backups/net-kingdom-identity',
                artifact='net-kingdom-identity-20261016T063159Z-5882771aa2ce.tar.age',
                ciphertext_bytes=32, ciphertext_sha256=hashlib.sha256(b'A' * 32).hexdigest(),
                etag='"original"')


@pytest.mark.parametrize('mode', ['account', 'folder-share'])
def test_offline_success_uses_only_fixed_provider_reads(record, mode):
    provider = Provider(b'A' * 32)
    record['untrusted_extra'] = 'must-not-export'
    result = m.retrieve(record, 'fixtureSecret1234', mode, provider)
    try:
        assert Path(result['fetched_path']).read_bytes() == b'A' * 32
        assert not result['openbao_contacted'] and not result['sso_contacted']
        assert {c[0] for c in provider.calls} == {'HEAD', 'GET'}
        assert 'fixtureSecret1234' not in json.dumps(result)
        assert 'must-not-export' not in json.dumps(result)
        assert provider.calls[1][2]['If-match'] == record['etag']
    finally:
        path = Path(result['fetched_path'])
        path.unlink()
        path.parent.rmdir()


@pytest.mark.parametrize('field,value', [
    ('artifact', '../another.tar.age'), ('folder', 'other'),
    ('ciphertext_sha256', 'x' * 64), ('ciphertext_bytes', True),
    ('provider', 'other'), ('etag', 'W/"weak"'),
])
def test_unreviewed_scope_refuses_before_contact(record, field, value):
    record[field] = value
    provider = Provider(b'A' * 32)
    with pytest.raises(m.offsite.Refused):
        m.retrieve(record, 'fixtureSecret1234', 'account', provider)
    assert not provider.calls


def test_digest_failure_removes_incomplete_ciphertext(record, monkeypatch, tmp_path):
    directory = tmp_path / 'download'
    directory.mkdir()
    monkeypatch.setattr(m.tempfile, 'mkdtemp', lambda **kwargs: str(directory))
    provider = Provider(b'A' * 32)
    provider.tamper = True
    with pytest.raises(m.offsite.Refused):
        m.retrieve(record, 'fixtureSecret1234', 'account', provider)
    assert not directory.exists()


def test_changed_etag_refuses_before_download(record):
    provider = Provider(b'A' * 32)
    provider.etag = '"changed"'
    with pytest.raises(m.offsite.Refused):
        m.retrieve(record, 'fixtureSecret1234', 'account', provider)
    assert len(provider.calls) == 1


def test_kit_contains_all_offline_imports_and_checksums(tmp_path):
    files = k.build()
    archive = next(value for name, value in files.items() if name.endswith('.zip'))
    with zipfile.ZipFile(io.BytesIO(archive)) as kit:
        kit.extractall(tmp_path)
        for line in kit.read('SHA256SUMS').decode().splitlines():
            digest, name = line.split('  ', 1)
            assert hashlib.sha256((tmp_path / name).read_bytes()).hexdigest() == digest
    for script in ('verify-identity.py', 'retrieve-identity-offline.py'):
        result = subprocess.run([sys.executable, str(tmp_path / 'tools/bootstrap-recovery' / script), '--help'],
                                capture_output=True)
        assert result.returncode == 0, result.stderr
    assert k.build() == files


def test_kit_publication_conditional_and_exact_readback():
    provider = Provider(b'')
    # Ensure initial HEAD returns 404 instead of the download-fixture response.
    provider.objects['unrelated'] = b'preserved'
    files = k.build()
    rows = k.publish(files, {'Authorization': 'fixture'}, provider)
    assert len(rows) == 3 and all(row['readback_verified'] for row in rows)
    assert len([c for c in provider.calls if c[0] == 'PUT']) == 3
    k.publish(files, {'Authorization': 'fixture'}, provider)
    assert len([c for c in provider.calls if c[0] == 'PUT']) == 3
    assert provider.objects['unrelated'] == b'preserved'


def test_kit_conflicting_existing_object_never_overwritten():
    provider = Provider(b'')
    files = k.build()
    first = next(iter(files))
    provider.objects[m.ACCOUNT_BASE + first] = b'X' * len(files[first])
    with pytest.raises(k.offsite.Refused):
        k.publish(files, {'Authorization': 'fixture'}, provider)
    assert not any(c[0] == 'PUT' for c in provider.calls)