NK-WP-0055-T13: kingpath-offsite --verify restores and checks the break-glass copy

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

@@ -125,3 +125,55 @@ 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/'
+
+
+KUBECONFIG=b'apiVersion: v1\nclusters: [{name: c, cluster: {server: "https://127.0.0.1:1"}}]\nusers: [{name: u, user: {}}]\n'
+SSHKEY=b'-----BEGIN OPENSSH PRIVATE KEY-----\nsynthetic\n-----END OPENSSH PRIVATE KEY-----\n'
+
+
+def uploaded(tmp_path,key,share):
+    items=[f'admin-kubeconfig={private(tmp_path,"kc",KUBECONFIG)}',f'tunnel-ssh-key={private(tmp_path,"ssh",SSHKEY)}']
+    return k.run(items,'pw',tmp_path/'upload.json',share,key[1],NOW)
+
+
+def keyfd(path):return os.open(path,os.O_RDONLY)
+
+
+def test_verify_fetches_decrypts_in_memory_and_checks_shapes(tmp_path,key):
+    share=Share();upload=uploaded(tmp_path,key,share);fd=keyfd(key[0])
+    try:record=k.verify(upload,'pw',fd,tmp_path/'verify.json',share)
+    finally:os.close(fd)
+    assert record['status']=='verified' and record['unrelated_key_refused'] and record['custody_key_matches_recipient']
+    assert record['items']=={'admin-kubeconfig':{'bytes':len(KUBECONFIG),'shape':'kubeconfig'},
+                             'tunnel-ssh-key':{'bytes':len(SSHKEY),'shape':'ssh-private-key'}}
+    text=(tmp_path/'verify.json').read_text();assert 'synthetic' not in text and '127.0.0.1' not in text
+
+
+def test_verify_refuses_a_custody_key_for_another_recipient(tmp_path,key):
+    share=Share();upload=uploaded(tmp_path,key,share)
+    other=tmp_path/'other.key';subprocess.run(['age-keygen','-o',str(other)],check=True,capture_output=True)
+    fd=keyfd(other)
+    try:
+        with pytest.raises(k.Refused):k.verify(upload,'pw',fd,tmp_path/'verify.json',share)
+    finally:os.close(fd)
+    assert [c[0] for c in share.calls].count('GET')==1  # only the upload readback; no fetch with the wrong key
+
+
+@pytest.mark.parametrize('change',['etag','sha','items','artifact'])
+def test_verify_refuses_a_changed_record_or_object(tmp_path,key,change):
+    share=Share();upload=uploaded(tmp_path,key,share);fd=keyfd(key[0])
+    if change=='etag':upload['etag']='"other"'
+    if change=='sha':upload['ciphertext_sha256']='0'*64
+    if change=='items':upload['items']={'admin-kubeconfig':1}
+    if change=='artifact':upload['artifact']='../escape.tar.age'
+    try:
+        with pytest.raises(k.Refused):k.verify(upload,'pw',fd,tmp_path/'verify.json',share)
+    finally:os.close(fd)
+    assert not (tmp_path/'verify.json').exists()
+
+
+def test_verify_refuses_tampered_ciphertext(tmp_path,key):
+    share=Share();upload=uploaded(tmp_path,key,share);share.tamper=True;fd=keyfd(key[0])
+    try:
+        with pytest.raises(k.Refused):k.verify(upload,'pw',fd,tmp_path/'verify.json',share)
+    finally:os.close(fd)