net-kingdom/bounded-executor/window_watch.py

#!/usr/bin/env python3
"""Window watch: the write role must not outlive a run (NK-WP-0058-T06; contract row O5).

  python3 bounded-executor/window_watch.py --mode flag|close [--openbao-addr URL] [--jwt-file FILE]

Rule: the write role (CCR-2026-0058) exists, no bounded-executor run Job is active in sso, and none was created in
the last GRACE seconds. A role in that state is a window left open, as on 2026-11-04/05. Run Jobs carry
app.kubernetes.io/name=bounded-executor and no app.kubernetes.io/component label (request, preflight, journey and
netcheck pods carry one).

  flag   read-only: print `flagged_stale` and exit 3, so the watch Job fails and its status shows it
  close  delete the role, prove it is absent, print `closed_stale`, exit 0. Closing only reduces privilege: a run
         that was about to start is refused before any effect.

The identity is ServiceAccount sso/nk-window-watch. It logs in to OpenBao role
net-kingdom-bounded-executor-window-watch, whose policy allows read (flag), or read and delete (close), on the
write role only, and it may list Jobs in sso. Prints one value-free line; exit 0 ok or closed, 3 flagged, 5 error.
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import sys
import urllib.parse
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import operations as OPS  # noqa: E402

WRITE_ROLE = 'net-kingdom-bind-executor-svc-authelia-write'
WATCH_ROLE = 'net-kingdom-bounded-executor-window-watch'
ROLE_PATH = '/v1/auth/kubernetes/role/' + WRITE_ROLE
RUN_JOBS = 'app.kubernetes.io/name=bounded-executor,!app.kubernetes.io/component'
GRACE = 600  # seconds: the window opens moments before its run Job is created


def role_present(bao) -> bool:
    try:
        bao._call('GET', ROLE_PATH)
        return True
    except OPS.OperationError as error:
        if str(error).endswith(' 404'):
            return False
        raise


def run_jobs(kube, namespace='sso'):
    query = urllib.parse.urlencode({'labelSelector': RUN_JOBS})
    return kube.get(f'/apis/batch/v1/namespaces/{namespace}/jobs?{query}').get('items', [])


def decide(present: bool, jobs: list, now: dt.datetime, grace: int = GRACE) -> tuple[str, dict]:
    """The rule, without I/O: ('ok' | 'stale', value-free facts)."""
    active = sum(1 for j in jobs if (j.get('status') or {}).get('active'))
    created = [dt.datetime.fromisoformat(j['metadata']['creationTimestamp'].replace('Z', '+00:00')) for j in jobs]
    newest = max(created, default=None)
    age = int((now - newest).total_seconds()) if newest else None
    facts = {'write_role_present': present, 'run_jobs': len(jobs), 'active': active, 'newest_age_s': age}
    stale = present and not active and (age is None or age > grace)
    return ('stale' if stale else 'ok'), facts


def watch(bao, kube, mode: str, now: dt.datetime) -> tuple[dict, int]:
    verdict, facts = decide(role_present(bao), run_jobs(kube), now)
    if verdict == 'ok':
        return {'status': 'ok', **facts}, 0
    if mode == 'flag':
        return {'status': 'flagged_stale', **facts}, 3
    bao._call('DELETE', ROLE_PATH)
    if role_present(bao):
        return {'status': 'error', 'reason': 'write_role_still_present', **facts}, 5
    return {'status': 'closed_stale', **facts}, 0


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__.split('\n', 1)[0])
    p.add_argument('--mode', choices=['flag', 'close'], required=True)
    p.add_argument('--openbao-addr', default='http://openbao.openbao.svc:8200')
    p.add_argument('--jwt-file', default='/var/run/secrets/openbao/token')
    a = p.parse_args(argv)
    bao = None
    try:
        bao = OPS.OpenBao(a.openbao_addr, k8s_role=WATCH_ROLE, jwt_file=a.jwt_file)
        result, code = watch(bao, OPS.Kube(), a.mode, dt.datetime.now(dt.timezone.utc))
    except Exception as error:  # value-free: the clients' own errors carry fixed texts
        reason = str(error) if isinstance(error, OPS.OperationError) else type(error).__name__
        result, code = {'status': 'error', 'reason': reason}, 5
    finally:
        if bao is not None:
            bao.revoke_self()
    print(json.dumps({'watch': 'bounded-execution-window', 'mode': a.mode, **result}, sort_keys=True))
    return code


if __name__ == '__main__':
    sys.exit(main())