kubestronaut-sim/sim/bank.py

"""Helpers shared by every track's task bank."""

NS = r"""
mkns() {
  $K create ns "$1" --dry-run=client -o yaml | $K apply -f - >/dev/null
  # Wait for the namespace's `default` ServiceAccount before returning. The controller creates it
  # asynchronously, and a Pod applied in the gap is rejected outright:
  #   error looking up service account <ns>/default: serviceaccount "default" not found
  # Setups that create a namespace and immediately apply Pods lose that race under load,
  # which fails the seed and costs the candidate a whole task. It belongs here, not in each task.
  # 90 s, not 30: the ServiceAccount controller runs in kube-controller-manager, and a manager that
  # is restarting (it takes ~60 s to be back and leading) left bare Pods rejected with no trace.
  for _ in $(seq 1 90); do $K -n "$1" get sa default >/dev/null 2>&1 && return 0; sleep 1; done
  echo "namespace $1 still has no default ServiceAccount after 90 s - is kube-controller-manager running?" >&2
  return 1
}
"""

def C(name, pts, script):
    return {"name": name, "pts": pts, "script": script}

def step(title, text, check, hint="", show=""):
    """One step of a lesson (<track>/lessons.py): what to do (Markdown), the check that proves it
    (bash, exit 0 = done, run like a task's checks), a hint, and how to do it as you would in the
    exam - shown on request, and done by ./ksim lesson audit to prove the step works as shown.
    `show` is commands you type as they stand, or a list of cmd() and hand() parts."""
    return {"title": title, "text": text, "check": check, "hint": hint, "show": show}

def cmd(text, how=""):
    """Commands you type as they stand, and what to look for (Markdown); the audit runs them as
    they stand."""
    return {"type": text, "how": how, "run": text}

def hand(type_, how, run):
    """What you do by hand: what you type to begin (`vim f`, `k edit ...`, `echo NAME > f`), what
    to do there (Markdown), and a script that does the same, which the audit runs instead - an
    editor needs a person, and so does a value read off the screen."""
    return {"type": type_, "how": how, "run": run}

def ns_pick(r, base):
    """Namespace name with a random suffix so the same task never lands on the same namespace twice."""
    return f"{base}-{r.choice(['alpha', 'bravo', 'delta', 'echo', 'kilo', 'nova', 'omega', 'zulu'])}"

def name_pick(r, base):
    return f"{base}-{r.randint(10, 99)}"


# ---- bash building blocks for setups and checks, shared by the tracks' banks ----------------
# Prepend what a script needs:   "setup": PODS + r"""...""",   C("...", 3, NET + r"""...""")

# Traffic probes for checks.
NET = r"""
# ok NS POD URL: one request from a Pod succeeds.   okr: for a path that should work, given a
# moment.   blocked: the request fails - at once, or once a policy applied a moment ago has landed.
ok() { $K -n "$1" exec "$2" -- wget -T 3 -qO- "$3" >/dev/null 2>&1; }
okr() { local i; for i in 1 2 3 4; do ok "$@" && return 0; sleep 2; done; return 1; }
blocked() { local i; for i in 1 2 3 4 5 6; do ok "$@" || return 0; sleep 3; done; return 1; }
# has -n NS KIND NAME: the object exists
has() { [ -n "$(j "$@" '{.metadata.name}')" ]; }
"""

# Scenario building blocks for setups (busybox Pods: each can be the client of a test as well).
PODS = NS + r"""
# srv NS NAME [LABELS]: a Pod that serves its own name on port 80, with a Service of its name. It
# has wget and nslookup too, so it can be the client of a test as well.
srv() {
  $K -n "$1" run "$2" --image=busybox:1.36 --port=80 --labels="${3:-app=$2}" --command -- \
    sh -c 'mkdir -p /www && echo "$HOSTNAME" > /www/index.html && exec httpd -f -p 80 -h /www' >/dev/null
  $K -n "$1" expose pod "$2" --port=80 >/dev/null
}
# cli NS NAME [LABELS]: a Pod to send requests from
cli() { $K -n "$1" run "$2" --image=busybox:1.36 --labels="${3:-app=$2}" --command -- sleep 86400 >/dev/null; }
ready() { $K -n "$1" wait --for=condition=Ready pod --all --timeout=240s >/dev/null; }
"""

# An object as JSON, for a Python expression.
PY = r"""
# py KUBECTL-GET-ARGS.. 'PYTHON': the object as `o` (a dict) for a Python expression or statements;
# the check passes when the last line, an expression, is true. Fails when the object is missing.
py() {
  local n=$#; local code="${!n}"
  $K get "${@:1:n-1}" -o json 2>/dev/null | CODE="$code" python3 -c '
import json, os, sys
try:
    o = json.load(sys.stdin)
except Exception:
    sys.exit(1)
src = os.environ["CODE"].strip().splitlines()
env = {"o": o, "json": json, "os": os}
exec("\n".join(src[:-1]), env)
sys.exit(0 if eval(src[-1], env) else 1)'
}
"""

# Permissions, asked the way a Role spells them.
# can SUBJECT-FLAGS.. -- VERB RESOURCE [-n NS]: kubectl auth can-i says yes.  cannot: it says no.
# A subresource is written here as in a Role - pods/exec - and handed to kubectl the way it wants
# it: to `can-i`, "pods/exec" is a Pod named exec, and the subresource is --subresource=exec.
CAN = r"""
_cani() {
  local out=() a
  for a in "$@"; do
    case "$a" in
      */exec|*/attach|*/log|*/portforward|*/proxy|*/status|*/scale|*/eviction|*/ephemeralcontainers|*/token) out+=("${a%/*}" "--subresource=${a##*/}") ;;
      *) out+=("$a") ;;
    esac
  done
  $K auth can-i "${out[@]}" 2>/dev/null | head -1
}
can() { local a=(); while [ "$1" != -- ]; do a+=("$1"); shift; done; shift; [ "$(_cani "$@" "${a[@]}")" = yes ]; }
cannot() { local a=(); while [ "$1" != -- ]; do a+=("$1"); shift; done; shift; [ "$(_cani "$@" "${a[@]}")" = no ]; }
"""