tegwick
1bbba59444
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10e2d-a5ef-7e92-b03d-b1694e915680
56 lines
2 KiB
Python
56 lines
2 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent attended admission of the fixed Authelia SMTP consumer.
|
|
|
|
Reuses the owner-verified telemetry mailbox without copying or rotating it.
|
|
Only the dedicated sso/authelia-smtp-eso role gains the exact read-only lane.
|
|
"""
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(ROOT))
|
|
spec = importlib.util.spec_from_file_location(
|
|
"authelia_smtp_owner_lane", ROOT / "telemetry_smtp_delivery.py",
|
|
)
|
|
lane = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(lane)
|
|
lane.LANE = dict(service_account="authelia-smtp-eso", namespace="sso",
|
|
role="authelia-smtp-eso", policy="authelia-smtp-eso",
|
|
kv_path="platform/data/workloads/railiance-telemetry/smtp")
|
|
lane.POLICY = ('path "' + lane.LANE["kv_path"] + '" { capabilities = ["read"] }\n'
|
|
'path "auth/token/lookup-self" { capabilities = ["read"] }\n'
|
|
'path "auth/token/revoke-self" { capabilities = ["update"] }\n')
|
|
|
|
|
|
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": "authelia.smtp-delivery.v1", "status": "failed"}
|
|
try:
|
|
lane.run(receipt)
|
|
except Exception as error:
|
|
receipt["reason"] = (
|
|
str(error) if isinstance(error, lane.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)
|