Prepare password-free inbound-library IMAP entry for attended custody
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Sibling of the telemetry SMTP entry for library@coulomb.social
(inbound-library ILIB-WP-0002-T07).

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1170398@bnt-lap001
Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
This commit is contained in:
codex 2026-10-05 11:14:54 +02:00
parent 6217eef2c4
commit 01e0576132
3 changed files with 162 additions and 0 deletions

View file

@ -0,0 +1,75 @@
# inbound-library IMAP entry
This is a sibling of the telemetry SMTP entry (`docs/telemetry-smtp-custody.md`).
It follows the same pattern and makes the same guarantees. The work belongs to
inbound-library ILIB-WP-0002-T07. The founder requested the entry on 2026-10-05
for the dedicated mailbox library@coulomb.social and will add the password.
KV v2 mount: `platform`. Entry: `workloads/inbound-library/imap`.
Full CLI path: `platform/workloads/inbound-library/imap`.
The initial fields are:
| Field | Value |
| --- | --- |
| `IMAP_HOST` | `imap.ionos.de` |
| `IMAP_PORT` | `993` |
| `IMAP_USERNAME` | `library@coulomb.social` |
| `IMAP_TLS` | `implicit` |
| `IMAP_FOLDER` | `INBOX` |
`IMAP_PASSWORD` is deliberately absent until the founder adds it as version 2.
Keep the existing fields when saving that new version, for example with
`bao kv patch platform/workloads/inbound-library/imap IMAP_PASSWORD=-`, which
reads the value from stdin.
This is an attended founder command. It needs the existing local OpenBao
forwarding:
```sh
BAO_ADDR=http://127.0.0.1:18200 VAULT_ADDR=http://127.0.0.1:18200 warden access openbao-platform-admin-login --exec -- python3 /home/worsch/railiance-platform/scripts/inbound_library_imap_entry.py
```
Behavior:
- The helper prints nothing.
- It writes with CAS=0, so it never overwrites any existing path or version.
- If a version already exists, it returns without reading values or writing.
- A successful rerun on an existing path proves only that the path exists, not
what it contains.
Four tests cover:
- first creation;
- an existing version is kept;
- permission errors;
- CAS conflict.
Exit codes:
| Code | Meaning |
| --- | --- |
| 20 | invalid metadata |
| 21 | metadata access failure |
| 22 | creation failure |
| 23 | unexpected write version |
| 24 | verification failure |
| 25 | contained failure |
Read Warden's printed completion line:
- `login-failed`: no child process ran.
- `completed-but-revocation-unconfirmed`: creation ran, but the session
revocation needs attention.
The native creation counts as pending until that attended result is in.
This helper does not do any of the following:
- grant read access to any workload;
- set a coding-agent deny boundary;
- copy another mailbox's credential;
- touch the mailbox.
inbound-library first runs on the workstation, with `INBOUND_LIBRARY_IMAP_USER`
and `INBOUND_LIBRARY_IMAP_PASSWORD` in its environment. Scoped delivery is a
separate, later step, done when the library moves into the cluster: a read
policy, an ESO ServiceAccount and an ExternalSecret, modelled on
`telemetry_smtp_delivery.py`. That later step also adds a certificate-verified,
read-only IMAP login check.

View file

@ -0,0 +1,55 @@
#!/usr/bin/env python3
"""Silent attended creation of a password-free inbound-library IMAP KV entry.
Never overwrite an existing version, including a password subsequently added by
its owner. Run inside warden's attended OpenBao platform-admin login envelope.
"""
import json
import os
import subprocess
import sys
PATH = 'platform/workloads/inbound-library/imap'
FIELDS = {'IMAP_HOST': 'imap.ionos.de', 'IMAP_PORT': '993',
'IMAP_USERNAME': 'library@coulomb.social',
'IMAP_TLS': 'implicit', 'IMAP_FOLDER': 'INBOX'}
def command(args):
return subprocess.run(['bao', *args], capture_output=True, timeout=30)
def run(invoke=command):
# Read only metadata before CAS=0 creation. No existing credential value is
# needed to refuse an overwrite. The owner can safely rerun after adding it.
before = invoke(['kv', 'metadata', 'get', '-format=json', PATH])
if before.returncode == 0:
metadata = json.loads(before.stdout)['data']
if type(metadata.get('current_version')) is int and metadata['current_version'] >= 1:
return 0
return 20
# Never infer absence from arbitrary errors or an unauthenticated connection.
if b'No value found at ' not in before.stderr + before.stdout:
return 21
created = invoke(['kv', 'put', '-format=json', '-cas=0', PATH,
*[key + '=' + value for key, value in FIELDS.items()]])
if created.returncode:
return 22
if json.loads(created.stdout).get('data', {}).get('version') != 1:
return 23
after = invoke(['kv', 'metadata', 'get', '-format=json', PATH])
if after.returncode or json.loads(after.stdout)['data'].get('current_version') != 1:
return 24
return 0
if __name__ == '__main__':
# The attended lane refuses all child output, including exceptions.
with open(os.devnull, 'w') as sink:
os.dup2(sink.fileno(), 1)
os.dup2(sink.fileno(), 2)
try:
status = run()
except Exception:
status = 25
sys.exit(status)

View file

@ -0,0 +1,32 @@
import importlib.util
import json
from pathlib import Path
import subprocess
import unittest
spec=importlib.util.spec_from_file_location('imap_entry',Path(__file__).resolve().parents[1]/'scripts/inbound_library_imap_entry.py')
entry=importlib.util.module_from_spec(spec);spec.loader.exec_module(entry)
class EntryTests(unittest.TestCase):
def test_create_without_password_and_verify(self):
calls=[]
results=[subprocess.CompletedProcess([],2,b'',b'No value found at platform/metadata/workloads/inbound-library/imap'),
subprocess.CompletedProcess([],0,b'{"data":{"version":1}}',b''),
subprocess.CompletedProcess([],0,b'{"data":{"current_version":1}}',b'')]
def invoke(args): calls.append(args);return results.pop(0)
self.assertEqual(entry.run(invoke),0)
self.assertIn('-cas=0',calls[1]);self.assertFalse(any('PASSWORD' in a for a in calls[1]))
self.assertIn('IMAP_USERNAME=library@coulomb.social',calls[1])
def test_existing_password_version_never_read_or_overwritten(self):
calls=[]
def invoke(args):
calls.append(args);return subprocess.CompletedProcess([],0,b'{"data":{"current_version":2}}',b'')
self.assertEqual(entry.run(invoke),0);self.assertEqual(len(calls),1)
self.assertEqual(calls[0][:3],['kv','metadata','get'])
def test_permission_error_never_triggers_write(self):
calls=[]
def invoke(args): calls.append(args);return subprocess.CompletedProcess([],2,b'',b'permission denied')
self.assertEqual(entry.run(invoke),21);self.assertEqual(len(calls),1)
def test_cas_conflict_is_not_success(self):
results=[subprocess.CompletedProcess([],2,b'',b'No value found at x'),subprocess.CompletedProcess([],2,b'',b'CAS mismatch')]
self.assertEqual(entry.run(lambda _:results.pop(0)),22)