Fix private recovery prompts on nonseekable terminals
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
This commit is contained in:
tegwick 2026-10-16 19:56:06 +02:00
parent b35e81f48f
commit fbdd62b7ed
5 changed files with 229 additions and 15 deletions

View file

@ -0,0 +1,85 @@
{
"schema": "netkingdom.identity-offline-terminal-fix.v1",
"date": "2026-10-16",
"original_attempt_receipt": {
"schema": "netkingdom.recovery-kit.v1",
"status": "failed",
"secret_values_exported": false,
"new_grants_created": false,
"files": [
{
"name": "net-kingdom-recovery-62b9464609b8e0fd-START-HERE.md",
"bytes": 8381,
"sha256": "1531675f2e7d3713a3a7fc28a5143fb24f201b19cd05d14c59ee485a0bc11e61"
},
{
"name": "net-kingdom-recovery-62b9464609b8e0fd.zip",
"bytes": 34527,
"sha256": "62b9464609b8e0fde2c05527fc79eff742ee3f626262d847df8fb579548a9470"
},
{
"name": "net-kingdom-recovery-62b9464609b8e0fd.zip.sha256",
"bytes": 108,
"sha256": "55e375291b27c2977049b89a5deddaed5e39c7ee9179013ce73d2ad43f2fb308"
}
],
"error": "kit_operation_failed_private_diagnostics_suppressed"
},
"reproduced_failure": "UnsupportedOperation: File or stream is not seekable opening a controlling terminal with Python text r+ before getpass",
"fix": "Shared getpass prompt with interactive stdin/stderr checks and GetPassWarning raised before echo fallback; no seek-requiring terminal open",
"tests": {
"command": "python3 -m pytest -q tests/test_identity_offline_recovery.py tests/test_identity_offsite.py",
"passed": 36,
"failed": 0,
"real_controlling_terminal_hidden_input_verified": true,
"redirected_input_refused": true,
"echo_fallback_refused": true,
"both_cli_failed_stage_receipts_verified": true
},
"sources": [
{
"path": "tools/bootstrap-recovery/retrieve-identity-offline.py",
"sha256": "3421dbb3f3280a001d33ea2deb12ab609404a0b4254cd43c21518d77d166f972"
},
{
"path": "tools/bootstrap-recovery/recovery-kit.py",
"sha256": "6fa0f1494dccaf003b6c6396c5de8b36faa2b276eeb16badd04383f5efb71e66"
},
{
"path": "tests/test_identity_offline_recovery.py",
"sha256": "827b4e135d6be7a54f3f56cfc9340828462a70866ffe3513b544256bf2dafa31"
}
],
"rebuilt_kit": {
"schema": "netkingdom.recovery-kit.v1",
"status": "built",
"secret_values_exported": false,
"new_grants_created": false,
"files": [
{
"name": "net-kingdom-recovery-57d1940947aa25ff-START-HERE.md",
"bytes": 8381,
"sha256": "1531675f2e7d3713a3a7fc28a5143fb24f201b19cd05d14c59ee485a0bc11e61"
},
{
"name": "net-kingdom-recovery-57d1940947aa25ff.zip",
"bytes": 34930,
"sha256": "57d1940947aa25ff83d0f25ddf30090eb40595cccaab64225de882a2fd59ea2f"
},
{
"name": "net-kingdom-recovery-57d1940947aa25ff.zip.sha256",
"bytes": 108,
"sha256": "102d1cd25662650a7a181039f6646883d72a9d976551a8950279d8936ca22171"
}
],
"finished_at": "2026-10-16T17:55:59.729885+00:00"
},
"actual_provider_publication_verified": false,
"actual_outage_retrieval_verified": false,
"private_credential_values_read_by_agent": false,
"knowledge": {
"version": "0.2.6",
"digest": "8d1d6a9d4d07fec7b619618de7dfe2be2254c1790f85f4198c7fe19db9751c02",
"status": "runbooks withheld; existing directly reviewed local owner contract governs scoped code repair"
}
}

View file

@ -6,6 +6,10 @@ import json
from pathlib import Path
import subprocess
import sys
import os
import pty
import select
import termios
import urllib.error
import zipfile
@ -25,6 +29,91 @@ m = load('retrieve-identity-offline')
k = load('recovery-kit')
def test_private_prompt_with_real_controlling_terminal_hides_input():
import fcntl
master, slave = pty.openpty()
def terminal_session():
os.setsid()
fcntl.ioctl(0, termios.TIOCSCTTY, 0)
code = ("import importlib.util; "
f"s=importlib.util.spec_from_file_location('offline',{str(ROOT / 'tools/bootstrap-recovery/retrieve-identity-offline.py')!r}); "
"m=importlib.util.module_from_spec(s); s.loader.exec_module(m); "
"value=m.private_prompt('Private fixture prompt: '); "
"print('accepted' if value=='fixture-never-echo-me' else 'rejected')")
child = subprocess.Popen([sys.executable, '-c', code], stdin=slave, stdout=slave,
stderr=slave, preexec_fn=terminal_session)
os.close(slave)
captured = b''
try:
assert select.select([master], [], [], 10)[0], 'prompt timeout'
while b'Private fixture prompt: ' not in captured:
captured += os.read(master, 4096)
assert select.select([master], [], [], 10)[0] or b'Private fixture prompt: ' in captured
os.write(master, b'fixture-never-echo-me\n')
child.wait(timeout=10)
while select.select([master], [], [], 0)[0]:
try:
block = os.read(master, 4096)
except OSError:
break
if not block:
break
captured += block
assert child.returncode == 0 and b'accepted' in captured
assert b'fixture-never-echo-me' not in captured
finally:
if child.poll() is None:
child.kill()
child.wait()
os.close(master)
def test_private_prompt_refuses_redirected_input(monkeypatch):
monkeypatch.setattr(m.sys.stdin, 'isatty', lambda: False)
monkeypatch.setattr(m.getpass, 'getpass', lambda *args: pytest.fail('must not prompt'))
with pytest.raises(m.offsite.Refused):
m.private_prompt('Private: ')
def test_private_prompt_never_falls_back_to_echo(monkeypatch):
import warnings
monkeypatch.setattr(m.sys.stdin, 'isatty', lambda: True)
monkeypatch.setattr(m.sys.stderr, 'isatty', lambda: True)
def cannot_hide(*args):
warnings.warn('fixture cannot hide input', m.getpass.GetPassWarning)
pytest.fail('must stop before fallback reads input')
monkeypatch.setattr(m.getpass, 'getpass', cannot_hide)
with pytest.raises(m.getpass.GetPassWarning):
m.private_prompt('Private: ')
def test_http_failure_receipt_metadata_never_includes_private_response():
error = urllib.error.HTTPError('https://private.invalid/secret', 401,
'private response', {'Authorization': 'secret'}, None)
assert m.failure_metadata(error) == {'failure_kind': 'provider_http_error', 'http_status': 401}
@pytest.mark.parametrize('script', ['retrieve-identity-offline.py', 'recovery-kit.py'])
def test_cli_refuses_unattended_prompt_with_safe_stage_receipt(tmp_path, script):
receipt = tmp_path / 'receipt.json'
args = [sys.executable, str(ROOT / 'tools/bootstrap-recovery' / script),
'--receipt', str(receipt)]
if script == 'recovery-kit.py':
args += ['--output-dir', str(tmp_path / 'kit'), '--publish']
else:
args += ['--source-receipt', str(ROOT / 'docs/evidence/2026-10-16-identity-backup-live/actual-offsite.json')]
result = subprocess.run(args, input=b'', capture_output=True, timeout=10)
assert result.returncode == 1
record = json.loads(receipt.read_text())
assert record['failed_stage'] == 'private_provider_prompt'
assert record['failure_kind'] == 'scope_or_integrity_refusal'
assert 'Traceback' not in result.stderr.decode()
class Response(io.BytesIO):
def __init__(self, url, raw, etag='"original"', status=200):
super().__init__(raw)

View file

@ -7,7 +7,6 @@ never changes shares, credentials, existing files, retention or runtime jobs.
import argparse
import base64
from datetime import datetime, timezone
import getpass
import hashlib
import importlib.util
import io
@ -81,7 +80,7 @@ def publish(files, headers, client=None):
present = True
except urllib.error.HTTPError as error:
if error.code != 404:
raise offsite.Refused() from None
raise
if not present:
h = dict(headers, **{'If-None-Match': '*', 'Content-Length': str(len(raw)),
'Content-Type': 'application/zip' if name.endswith('.zip') else 'text/plain; charset=utf-8'})
@ -114,10 +113,13 @@ def main():
fd = os.open(a.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
r = {'schema': 'netkingdom.recovery-kit.v1', 'status': 'failed',
'secret_values_exported': False, 'new_grants_created': False}
stage = 'option_validation'
try:
if a.offline_transfer_receipt and (not a.publish or a.auth_source != 'prompt'):
raise offsite.Refused()
stage = 'build_kit'
files = build()
stage = 'write_local_kit'
directory = Path(a.output_dir)
directory.mkdir(mode=0o700, exist_ok=False)
for name, raw in files.items():
@ -126,11 +128,10 @@ def main():
for n, v in files.items()])
if a.publish:
if a.auth_source == 'prompt':
with open('/dev/tty', 'r+') as terminal:
if not terminal.isatty():
raise offsite.Refused()
secret = getpass.getpass('Independently held Storage Share Backup app password: ', stream=terminal)
stage = 'private_provider_prompt'
secret = offline.private_prompt('Independently held Storage Share Backup app password: ')
else:
stage = 'attended_owner_custody'
spec = importlib.util.spec_from_file_location('inventory', HERE / 'inventory-storage-share.py')
owner = importlib.util.module_from_spec(spec)
spec.loader.exec_module(owner)
@ -141,10 +142,12 @@ def main():
try:
headers = {'Authorization': 'Basic ' + base64.b64encode(('Backup:' + secret).encode()).decode(),
'Accept-Encoding': 'identity'}
stage = 'provider_publication'
r.update(status='published_readback_verified', files=publish(files, headers),
folder='Backup/railiance-backups/net-kingdom-identity',
provider='Storage Share', credential_source=a.auth_source)
if a.offline_transfer_receipt:
stage = 'offline_provider_download'
transfer_fd = os.open(a.offline_transfer_receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
transfer = {'status': 'failed', 'error': 'offline_retrieval_failed_private_diagnostics_suppressed'}
try:
@ -158,8 +161,9 @@ def main():
finally:
secret = None
r['finished_at'] = datetime.now(timezone.utc).isoformat()
except (Exception, KeyboardInterrupt):
r.update(status='failed', error='kit_operation_failed_private_diagnostics_suppressed')
except (Exception, KeyboardInterrupt) as error:
r.update(status='failed', error='kit_operation_failed_private_diagnostics_suppressed',
failed_stage=stage, **offline.failure_metadata(error))
finally:
with os.fdopen(fd, 'w') as output:
json.dump(r, output, indent=2)

View file

@ -19,6 +19,8 @@ import signal
import sys
import tempfile
import urllib.request
import urllib.error
import warnings
HERE = Path(__file__).resolve().parent
spec = importlib.util.spec_from_file_location('offsite', HERE / 'identity-offsite.py')
@ -27,6 +29,30 @@ spec.loader.exec_module(offsite)
ACCOUNT_BASE = offsite.HOST + '/remote.php/dav/files/Backup/railiance-backups/net-kingdom-identity/'
def private_prompt(label):
# getpass opens the terminal with os.open, which does not require seeking.
# Reject redirected input and turn its echoing fallback warning into an
# exception before it can read a credential without echo suppression.
if not sys.stdin.isatty() or not sys.stderr.isatty():
raise offsite.Refused()
with warnings.catch_warnings():
warnings.simplefilter('error', getpass.GetPassWarning)
return getpass.getpass(label)
def failure_metadata(error):
# No exception messages, response bodies, URLs or headers enter receipts.
if isinstance(error, urllib.error.HTTPError):
return {'failure_kind': 'provider_http_error', 'http_status': int(error.code)}
if isinstance(error, urllib.error.URLError):
return {'failure_kind': 'provider_transport_error'}
if isinstance(error, getpass.GetPassWarning):
return {'failure_kind': 'terminal_echo_suppression_unavailable'}
if isinstance(error, offsite.Refused):
return {'failure_kind': 'scope_or_integrity_refusal'}
return {'failure_kind': 'local_operation_failed'}
def validate(record):
if (record.get('schema') != 'netkingdom.identity-offsite.v1'
or record.get('status') != 'readback_verified'
@ -108,20 +134,20 @@ def main():
os.umask(0o077)
fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
result = {'status': 'failed', 'error': 'offline_retrieval_failed_private_diagnostics_suppressed'}
stage = 'source_receipt_validation'
try:
record = json.loads(Path(args.source_receipt).read_text())
validate(record)
# getpass must never fall back to echoed stdin (e.g. a captured agent tool).
with open('/dev/tty', 'r+') as terminal:
if not terminal.isatty():
raise offsite.Refused()
secret = getpass.getpass('Storage Share Backup app password or exact folder-share token: ', stream=terminal)
stage = 'private_provider_prompt'
secret = private_prompt('Storage Share Backup app password or exact folder-share token: ')
try:
stage = 'provider_download'
result = retrieve(record, secret, args.auth)
finally:
secret = None
except (Exception, KeyboardInterrupt):
pass
except (Exception, KeyboardInterrupt) as error:
result.update(failed_stage=stage, **failure_metadata(error))
finally:
with os.fdopen(fd, 'w') as output:
json.dump(result, output, indent=2)

View file

@ -676,6 +676,14 @@ passed 65 tests including real disposable age/PostgreSQL fixtures. ZIP-contained
entry points import without sibling repositories. This is source/fixture proof,
not actual independent-provider retrieval (T23) or publication (T26).
2026-10-16 correction: the first owner handoff stopped before its credential
prompt because Python text `r+` requires a seekable stream. Reproduced against a
real controlling PTY; replaced both prompts with the shared hidden-input helper,
refusing redirected input and echo fallback. Added real-terminal and both-CLI
regressions plus closed stage/HTTP-code diagnostics (no raw responses). The
focused rerun passed 36 tests. Evidence: `terminal-prompt-fix.json` in the outage
evidence directory. This source correction does not complete T23/T26.
## Publish and verify the disaster recovery guide on Storage Share
```task
@ -702,7 +710,9 @@ access to the recovery scripts; retain an independent copy and trusted digest.
`recovery-kit.py --publish` command through the existing attended Backup-account
owner conduit. Existing publisher/retention workload grants are not repurposed.
The user authorized this documentation upload; no new grant is required by the
prepared command. Credentials remain private; no provider permissions are widened.
prepared command. The first owner run (`/tmp/nk49-offline-handoff-VKv337zo`)
failed before private input/publication; the terminal bug is fixed in T25 and
the rebuilt kit is ready for a fresh owner run. Credentials remain private; no provider permissions are widened.
**Done when:** the versioned START-HERE guide, self-contained ZIP and ZIP SHA256
file are present beside the identity backups with exact conditional-create and