Prepare attended scoped platform mailbox admission for Authelia
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10e2d-a5ef-7e92-b03d-b1694e915680
This commit is contained in:
parent
7e569e8c26
commit
1bbba59444
4 changed files with 213 additions and 0 deletions
56
scripts/authelia_smtp_delivery.py
Normal file
56
scripts/authelia_smtp_delivery.py
Normal file
|
|
@ -0,0 +1,56 @@
|
|||
#!/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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue