railiance-telemetry/scripts/alert_service.py

#!/usr/bin/env python3
"""Single-process private acknowledgment runtime, packaged by rapp-telemetry."""
import argparse
from http.cookies import SimpleCookie
import json
import os
from pathlib import Path
import secrets
import signal
import threading
import time
from urllib.parse import parse_qs

from alert_ack import Application, Store
from alert_audit import AuditTransport
from alert_identity import Login
from alert_policy import Policy


def credential(path):
    # Kubernetes projected credentials use symlinks; mount ownership is the
    # package trust boundary. Resolve once per call so rotations are picked up.
    with Path(path).open('rb') as source:
        raw = source.read(32769)
    value = raw.decode().strip()
    if not 32 <= len(value) <= 32768 or any(ord(c) < 33 or ord(c) > 126 for c in value):
        raise ValueError('invalid credential projection')
    return value


class Router:
    def __init__(self, config, store, *, login=None, policy=None):
        self.config, self.store = config, store
        self.login = login or Login(config['issuer'], config['origin'])
        self.policy = policy or Policy(config['policy']['origin'],
            lambda: credential(config['policy']['token_file']), config['policy']['package'],
            config['policy']['version'], config['policy']['digest'])
        self.application = Application(store, config['origin'], credential(config['webhook_token_file']),
                                       self.actor, self.policy)
        self.last_drain = 0

    @staticmethod
    def cookie(env, name):
        cookies = SimpleCookie()
        try:
            cookies.load(env.get('HTTP_COOKIE', ''))
            return cookies[name].value if name in cookies else ''
        except Exception:
            return ''

    def actor(self, env):
        return self.login.session(self.cookie(env, '__Host-rtel-session'))

    def __call__(self, env, start):
        def reply(status, text='', headers=()):
            start(status, [('Content-Type', 'text/plain; charset=utf-8'), ('Cache-Control', 'no-store'),
                           ('Referrer-Policy', 'no-referrer'), ('X-Content-Type-Options', 'nosniff'),
                           ('Content-Security-Policy', "default-src 'none'; frame-ancestors 'none'")] + list(headers))
            return [text.encode()]
        def cookie(name, value, age):
            return ('Set-Cookie', f'{name}={value}; Path=/; Secure; HttpOnly; SameSite=Lax; Max-Age={age}')
        path, method = env.get('PATH_INFO'), env.get('REQUEST_METHOD')
        try:
            if method == 'GET' and path in ('/healthz', '/readyz'):
                ready = path == '/healthz' or time.time() - self.last_drain < 90 and not self.store.audit_debt()
                return reply('200 OK' if ready else '503 Service Unavailable', 'ready' if ready else 'audit delivery pending')
            if path == '/ack/auth/callback' and method == 'GET':
                params = parse_qs(env.get('QUERY_STRING', ''), strict_parsing=True, max_num_fields=4)
                if any(len(v) != 1 for v in params.values()) or not {'state', 'code'} <= set(params):
                    raise ValueError('invalid callback')
                sid, target = self.login.finish(params['state'][0], self.cookie(env, '__Host-rtel-login'), params['code'][0])
                return reply('303 See Other', headers=[('Location', target), cookie('__Host-rtel-session', sid, 900),
                                                       cookie('__Host-rtel-login', '', 0)])
            if path == '/ack/logout' and method == 'POST':
                actor = self.actor(env)
                form = parse_qs(Application.body(env).decode(), max_num_fields=1, strict_parsing=True)
                if (not actor or env.get('HTTP_ORIGIN') != self.config['origin'] or set(form) != {'csrf'}
                        or len(form['csrf']) != 1 or not secrets.compare_digest(actor.csrf, form['csrf'][0])):
                    return reply('403 Forbidden', 'Invalid sign-out.')
                self.login.logout(self.cookie(env, '__Host-rtel-session'))
                return reply('200 OK', 'Signed out.', [cookie('__Host-rtel-session', '', 0)])
            if path == '/ack/alerts' and method == 'GET' and self.actor(env) is None:
                target = path + '?' + env.get('QUERY_STRING', '')
                url, browser = self.login.start(target)
                return reply('303 See Other', headers=[('Location', url), cookie('__Host-rtel-login', browser, 300)])
            # Refresh the webhook credential for projected rotation; never store it in SQLite.
            if path == '/webhook':
                self.application.webhook_token = credential(self.config['webhook_token_file'])
            return self.application(env, start)
        except (ValueError, OSError, KeyError, TypeError):
            return reply('503 Service Unavailable', 'Identity or service unavailable. Retry later.')


def main():
    from waitress import serve
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--config', required=True, type=Path)
    args = parser.parse_args()
    os.umask(0o077)
    config = json.loads(args.config.read_text())
    Path(config['database']).parent.mkdir(mode=0o700, exist_ok=True)
    store = Store(config['database'])
    router = Router(config, store)
    transport = AuditTransport(config['audit']['origin'], lambda: credential(config['audit']['token_file']),
                               allow_internal_http=True)
    stop = threading.Event()
    def drain():
        while not stop.is_set():
            try:
                store.drain(transport)
                router.last_drain = time.time()
            except (OSError, ValueError):
                router.last_drain = 0
            stop.wait(30)
    worker = threading.Thread(target=drain, daemon=True)
    worker.start()
    def shutdown(*args):
        raise SystemExit(0)
    signal.signal(signal.SIGTERM, shutdown)
    try:
        serve(router, host='0.0.0.0', port=8080, threads=4, max_request_body_size=32768,
              clear_untrusted_proxy_headers=True)
    finally:
        stop.set()
        worker.join(timeout=6)


if __name__ == '__main__':
    main()