56fcc4c921
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f
118 lines
5 KiB
Python
118 lines
5 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent attended SMTP custody verification and exact ESO reader admission.
|
|
|
|
Never rewrites mailbox data. No Kubernetes Secret reads. Receipt contains only
|
|
fixed status fields and KV version. Does not send email or activate Alertmanager.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import smtplib
|
|
import ssl
|
|
import sys
|
|
|
|
from state_hub_preflight_lane import bao, data, LaneError, capabilities, revoke
|
|
from repair_eso_kubernetes_auth import role_payload, check_role
|
|
from telemetry_smtp_entry import FIELDS
|
|
|
|
LANE = dict(service_account='telemetry-smtp-eso', namespace='telemetry',
|
|
role='telemetry-smtp-eso', policy='telemetry-smtp-eso',
|
|
kv_path='platform/data/workloads/railiance-telemetry/smtp')
|
|
POLICY = ('path "'+LANE['kv_path']+'" { capabilities = ["read"] }\n'
|
|
'path "auth/token/lookup-self" { capabilities = ["read"] }\n'
|
|
'path "auth/token/revoke-self" { capabilities = ["update"] }\n')
|
|
|
|
|
|
def require(ok, reason):
|
|
if not ok:
|
|
raise LaneError(reason)
|
|
|
|
|
|
def validate_entry(native):
|
|
values = native['data']
|
|
require(all(values.get(k) == v for k,v in FIELDS.items()), 'smtp_settings_differ')
|
|
password = values.get('SMTP_PASSWORD')
|
|
require(isinstance(password, str) and bool(password.strip()) and len(password) <= 4096
|
|
and '\r' not in password and '\n' not in password, 'smtp_password_missing_or_invalid')
|
|
return password
|
|
|
|
|
|
def missing(result):
|
|
return result.returncode != 0 and b'No value found' in result.stdout + result.stderr
|
|
|
|
|
|
def ensure(path, wanted, check):
|
|
old = bao(['read','-format=json',path], allow_failure=True)
|
|
if old.returncode == 0:
|
|
check(data(old)['data'])
|
|
else:
|
|
require(missing(old), 'metadata_absence_unproven')
|
|
bao(['write',path,'-'], payload=wanted)
|
|
check(data(bao(['read','-format=json',path]))['data'])
|
|
|
|
|
|
def run(receipt):
|
|
identity = data(bao(['token','lookup','-format=json']))['data']['policies']
|
|
require('platform-admin' in identity and 'root' not in identity, 'attended_platform_admin_required')
|
|
native = data(bao(['read','-format=json',LANE['kv_path']]))['data']
|
|
password = validate_entry(native)
|
|
# Exact fixed endpoint; standard certificate validation and STARTTLS mandatory.
|
|
with smtplib.SMTP('smtp.ionos.de',587,timeout=15) as smtp:
|
|
smtp.ehlo()
|
|
smtp.starttls(context=ssl.create_default_context())
|
|
smtp.ehlo()
|
|
smtp.login(FIELDS['SMTP_USERNAME'], password)
|
|
receipt.update(kv_version=native['metadata']['version'], smtp_authenticated=True)
|
|
boundary = 'sys/policies/acl/agent-high-risk-boundary'
|
|
current = data(bao(['read','-format=json',boundary]))['data']['policy']
|
|
added = ''
|
|
for path in [LANE['kv_path'], LANE['kv_path'].replace('/data/','/metadata/',1)]:
|
|
stanza = 'path "'+path+'" { capabilities = ["deny"] }\n'
|
|
if '"'+path+'"' not in current:
|
|
added += stanza
|
|
else:
|
|
require(re.search(r'path\s+"'+re.escape(path)+r'"\s*\{\s*capabilities\s*=\s*\["deny"\]\s*\}',current), 'boundary_drift')
|
|
if added:
|
|
require(data(bao(['read','-format=json',boundary]))['data']['policy'] == current,'boundary_changed')
|
|
bao(['write',boundary,'-'],payload={'policy':current+'\n'+added})
|
|
require(data(bao(['read','-format=json',boundary]))['data']['policy'] == current+'\n'+added,'boundary_readback_failed')
|
|
ensure('sys/policies/acl/'+LANE['policy'], {'policy':POLICY},
|
|
lambda actual: require(actual['policy'] == POLICY,'reader_policy_drift'))
|
|
ensure('auth/kubernetes/role/'+LANE['role'],role_payload(LANE),lambda actual: check_role(actual,LANE))
|
|
child = data(bao(['token','create','-format=json','-policy='+LANE['policy'],
|
|
'-policy=agent-high-risk-boundary','-no-default-policy','-ttl=60s']))['auth']['client_token']
|
|
try:
|
|
paths = [LANE['kv_path'],LANE['kv_path'].replace('/data/','/metadata/',1)]
|
|
require(all(v == ['deny'] for v in capabilities(child,paths).values()),'agent_boundary_failed')
|
|
finally:
|
|
revoke(child)
|
|
receipt.update(status='verified', reader_admitted=True, coding_agent_denied=True,
|
|
password_modified=False, email_sent=False)
|
|
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--receipt',required=True,type=Path)
|
|
args=parser.parse_args()
|
|
fd=os.open(args.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
|
|
receipt={'schema':'telemetry.smtp-delivery.v1','status':'failed'}
|
|
try:
|
|
run(receipt)
|
|
except Exception as exc:
|
|
receipt['reason']=str(exc) if isinstance(exc,LaneError) else 'contained_operation_failed'
|
|
finally:
|
|
with os.fdopen(fd,'w') as target:
|
|
json.dump(receipt,target,indent=2);target.write('\n')
|
|
return 0 if receipt['status']=='verified' else 1
|
|
|
|
|
|
if __name__=='__main__':
|
|
with open(os.devnull,'w') as sink:
|
|
os.dup2(sink.fileno(),1);os.dup2(sink.fileno(),2)
|
|
try:
|
|
code=main()
|
|
except Exception:
|
|
code=1
|
|
sys.exit(code)
|