NK-WP-0055-T13: kingpath-offsite helper seals break-glass material to Storage Share

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 4515@bnt-lap001
Assistant-Session: 0bf96127-6e04-41be-bfda-d11717c3f354

tests/test_kingpath_offsite.py

@@ -0,0 +1,127 @@
+"""Break-glass bundle sealing and Storage Share publication on contained fixtures."""
+from datetime import datetime,timezone
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import tarfile
+import urllib.error
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+spec=importlib.util.spec_from_file_location('kingpath_offsite',ROOT/'tools/bootstrap-recovery/kingpath-offsite.py')
+k=importlib.util.module_from_spec(spec);spec.loader.exec_module(k)
+NOW=datetime(2026,10,27,22,0,tzinfo=timezone.utc)
+
+
+class Response(io.BytesIO):
+    def __init__(self,url,data=b'',status=200,etag='"v1"'):
+        super().__init__(data);self.url=url;self.status=status
+        self.headers={'ETag':etag,'Content-Length':str(len(data))}
+    def __enter__(self):return self
+    def __exit__(self,*a):self.close()
+
+
+class Share:
+    def __init__(self,folder_exists=False):
+        self.objects={};self.folder=folder_exists;self.calls=[];self.tamper=False;self.collide=False
+    def open(self,request,timeout=None):
+        method=request.get_method();url=request.full_url;self.calls.append((method,url))
+        assert request.headers['Authorization'].startswith('Basic ')
+        if method=='MKCOL':
+            if self.folder:raise urllib.error.HTTPError(url,405,'exists',{},None)
+            self.folder=True;return Response(url,status=201)
+        if method=='PUT':
+            assert request.headers['If-none-match']=='*'
+            self.objects[url]=request.data.read();return Response(url,status=201)
+        if self.collide and method=='HEAD':return _head(url,b'other')
+        if url not in self.objects:raise urllib.error.HTTPError(url,404,'missing',{},None)
+        data=self.objects[url]
+        if method=='HEAD':return _head(url,data)
+        if self.tamper:data=data[:-1]+bytes([data[-1]^1])
+        return Response(url,data)
+
+
+def _head(url,data):
+    r=Response(url,b'');r.headers['Content-Length']=str(len(data));return r
+
+
+@pytest.fixture
+def key(tmp_path):
+    identity=tmp_path/'fixture.key'
+    subprocess.run(['age-keygen','-o',str(identity)],check=True,capture_output=True)
+    recipient=subprocess.run(['age-keygen','-y',str(identity)],check=True,capture_output=True,text=True).stdout.strip()
+    return identity,recipient
+
+
+def private(tmp_path,name,data,mode=0o600):
+    path=tmp_path/name;path.write_bytes(data);path.chmod(mode);return path
+
+
+def test_seals_uploads_and_reads_back(tmp_path,key):
+    identity,recipient=key;share=Share()
+    a=private(tmp_path,'kc',b'synthetic-kubeconfig');b=private(tmp_path,'ssh',b'synthetic-ssh-key')
+    receipt=tmp_path/'receipt.json'
+    record=k.run([f'kubeconfig={a}',f'tunnel-ssh-key={b}'],'synthetic-app-password',receipt,share,recipient,NOW)
+    assert record['status']=='readback_verified' and record['items']=={'kubeconfig':20,'tunnel-ssh-key':17}
+    assert record['artifact'].startswith('net-kingdom-kingpath-20261027T220000Z-')
+    assert oct(receipt.stat().st_mode&0o777)=='0o600'
+    text=receipt.read_text();assert 'synthetic' not in text
+    (stored,)=share.objects.values();assert stored.startswith(b'age-encryption.org/v1\n') and b'synthetic' not in stored
+    plain=subprocess.run(['age','-d','-i',str(identity)],input=stored,capture_output=True,check=True).stdout
+    with tarfile.open(fileobj=io.BytesIO(plain)) as archive:
+        assert {m.name:(archive.extractfile(m).read(),m.mtime,m.mode) for m in archive}=={
+            'kubeconfig':(b'synthetic-kubeconfig',0,0o600),'tunnel-ssh-key':(b'synthetic-ssh-key',0,0o600)}
+    assert [c[0] for c in share.calls]==['MKCOL','HEAD','PUT','HEAD','GET']
+
+
+def test_existing_folder_is_accepted(tmp_path,key):
+    share=Share(folder_exists=True)
+    k.run([f'kubeconfig={private(tmp_path,"kc",b"x")}'],'pw',tmp_path/'r.json',share,key[1],NOW)
+    assert len(share.objects)==1
+
+
+@pytest.mark.parametrize('mode',[0o640,0o604,0o644])
+def test_refuses_group_or_world_readable_input(tmp_path,key,mode):
+    with pytest.raises(k.Refused):
+        k.run([f'kubeconfig={private(tmp_path,"kc",b"x",mode)}'],'pw',tmp_path/'r.json',Share(),key[1],NOW)
+
+
+def test_refuses_symlink_bad_name_duplicate_and_empty(tmp_path,key):
+    real=private(tmp_path,'kc',b'x');link=tmp_path/'link';link.symlink_to(real)
+    for specs in ([f'kubeconfig={link}'],[f'Bad_Name={real}'],[f'a={real}',f'a={real}'],
+                  [f'a={private(tmp_path,"empty",b"")}'],[],[f'a={tmp_path/"missing"}']):
+        with pytest.raises(k.Refused):k.run(specs,'pw',tmp_path/'r.json',Share(),key[1],NOW)
+
+
+def test_refuses_existing_receipt_before_any_provider_call(tmp_path,key):
+    share=Share();receipt=private(tmp_path,'r.json',b'{}')
+    with pytest.raises(k.Refused):k.run([f'a={private(tmp_path,"kc",b"x")}'],'pw',receipt,share,key[1],NOW)
+    assert share.calls==[]
+
+
+def test_refuses_tampered_readback_and_writes_no_receipt(tmp_path,key):
+    share=Share();share.tamper=True;receipt=tmp_path/'r.json'
+    with pytest.raises(k.Refused):k.run([f'a={private(tmp_path,"kc",b"x")}'],'pw',receipt,share,key[1],NOW)
+    assert not receipt.exists()
+
+
+def test_refuses_overwriting_an_existing_object(tmp_path,key):
+    share=Share();share.collide=True
+    with pytest.raises(k.Refused):k.run([f'a={private(tmp_path,"kc",b"x")}'],'pw',tmp_path/'r.json',share,key[1],NOW)
+    assert 'PUT' not in [c[0] for c in share.calls]
+
+
+def test_refuses_bad_credential_and_recipient(tmp_path,key):
+    item=private(tmp_path,'kc',b'x')
+    for pw in ('','a\nb'):
+        with pytest.raises(k.Refused):k.run([f'a={item}'],pw,tmp_path/'r.json',Share(),key[1],NOW)
+    with pytest.raises(k.Refused):k.run([f'a={item}'],'pw',tmp_path/'r.json',Share(),'not-a-recipient',NOW)
+
+
+def test_production_recipient_is_the_registered_backup_recipient():
+    profile=json.loads((ROOT/'tools/bootstrap-recovery/identity-backup-profile.json').read_text())
+    assert k.RECIPIENT==profile['public_recipient']
+    assert k.ACCOUNT_BASE=='https://nx4069.your-storageshare.de/remote.php/dav/files/Backup/railiance-backups/net-kingdom-kingpath/'