diff --git a/docs/inbound-library-imap-custody.md b/docs/inbound-library-imap-custody.md new file mode 100644 index 0000000..d86f7b5 --- /dev/null +++ b/docs/inbound-library-imap-custody.md @@ -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. diff --git a/scripts/inbound_library_imap_entry.py b/scripts/inbound_library_imap_entry.py new file mode 100644 index 0000000..15b15be --- /dev/null +++ b/scripts/inbound_library_imap_entry.py @@ -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) diff --git a/tests/test_inbound_library_imap_entry.py b/tests/test_inbound_library_imap_entry.py new file mode 100644 index 0000000..808f714 --- /dev/null +++ b/tests/test_inbound_library_imap_entry.py @@ -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)