railiance-platform/scripts/kube_route.py
codex 1228cd8407
All checks were successful
CCR intake gate / gate (push) Successful in 2m11s
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
RPF-WP-0044-T06: founder cluster route and attended ESO adoption script
KUBECTL_ROUTE (agent default, founder-sudo = ssh railiance01 sudo -n) for the
bounded proof scripts; one attended script with go/no-go gates, typed ADOPT
confirmation, prune/force-off SSA sync and post-checks; evaluator and tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 527080@bnt-lap001
Assistant-Session: 5a0e864a-67be-4605-a77a-16635f7dd292
2026-10-21 01:23:16 +02:00

38 lines
1.4 KiB
Python

"""One configurable cluster route for bounded railiance01 proof scripts.
KUBECTL_ROUTE selects how a remote command reaches the railiance01 API:
* ``agent`` (default): ``ssh railiance01 <command>`` as the SSH login itself.
Since the CUST-WP-0073 cutover the agent login has no kubeconfig, so this
route fails closed for agents instead of silently escalating.
* ``founder-sudo``: ``ssh railiance01 sudo -n <command>``, the founder's
attended admin path (root reads the k3s kubeconfig). ``-n`` never prompts,
so stdin/stdout stay byte-clean pipes (no ``ssh -t``, which would merge
stderr into stdout and rewrite line endings).
Only these named routes exist; arbitrary shell is never accepted.
"""
import os
import shlex
SSH = ['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=10', 'railiance01']
ROUTES = {'agent': [], 'founder-sudo': ['sudo', '-n']}
def route_name(environ=None):
name = (os.environ if environ is None else environ).get('KUBECTL_ROUTE', 'agent')
if name not in ROUTES:
raise ValueError('unknown KUBECTL_ROUTE')
return name
def remote_prefix(environ=None):
return list(ROUTES[route_name(environ)])
def ssh_argv(remote_args, environ=None):
"""Local argv that runs ``remote_args`` on railiance01 through the selected route."""
remote_args = [str(a) for a in remote_args]
if not remote_args:
raise ValueError('remote command required')
return SSH + [shlex.join(remote_prefix(environ) + remote_args)]