116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import re
|
|
import sys
|
|
import tarfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def read_text(zf: zipfile.ZipFile, name: str) -> str:
|
|
return zf.read(name).decode("utf-8")
|
|
|
|
|
|
def read_json(zf: zipfile.ZipFile, name: str) -> Any:
|
|
return json.loads(read_text(zf, name))
|
|
|
|
|
|
def read_jsonl(zf: zipfile.ZipFile, name: str) -> list[dict[str, Any]]:
|
|
rows = []
|
|
for line in read_text(zf, name).splitlines():
|
|
line = line.strip()
|
|
if line:
|
|
rows.append(json.loads(line))
|
|
return rows
|
|
|
|
|
|
def extract_snapshot_text(zf: zipfile.ZipFile, review_short: str) -> str:
|
|
blob = zf.read("ingress-nginx/config_snapshots.tar")
|
|
with tarfile.open(fileobj=io.BytesIO(blob), mode="r") as tf:
|
|
member = tf.extractfile(f"snapshots/review-{review_short}/nginx.conf")
|
|
if member is None:
|
|
raise ValueError("missing matching nginx config snapshot")
|
|
return member.read().decode("utf-8")
|
|
|
|
|
|
def recover_components(bundle_path: str | Path) -> dict[str, str]:
|
|
with zipfile.ZipFile(bundle_path) as zf:
|
|
access_rows = read_jsonl(zf, "network/admission_service_access.jsonl")
|
|
direct = next(
|
|
row
|
|
for row in access_rows
|
|
if row.get("method") == "POST"
|
|
and row.get("source_category") == "workload-pod"
|
|
and not row.get("apiserver_proxy")
|
|
)
|
|
review_uid = direct["review_uid"]
|
|
review_short = review_uid.split("-", 1)[0][:6]
|
|
|
|
review = read_json(zf, direct["review_file"])
|
|
if review["request"]["uid"] != review_uid:
|
|
raise ValueError("AdmissionReview file does not match access log UID")
|
|
annotations = review["request"]["object"]["metadata"].get("annotations", {})
|
|
if "validation-template" not in " ".join(annotations):
|
|
raise ValueError("matched review lacks validation-template annotation")
|
|
|
|
controller_log = read_text(zf, "ingress-nginx/controller.log")
|
|
if review_uid not in controller_log:
|
|
raise ValueError("review UID missing from controller log")
|
|
|
|
snapshot = extract_snapshot_text(zf, review_short)
|
|
proc_fd_match = re.search(r"/proc/\d+/fd/\d+", snapshot)
|
|
if not proc_fd_match:
|
|
raise ValueError("matching snapshot does not contain a proc fd marker")
|
|
|
|
falco_rows = read_jsonl(zf, "runtime/falco_events.jsonl")
|
|
runtime = next(
|
|
row
|
|
for row in falco_rows
|
|
if row.get("review_uid") == review_uid and row.get("fd.name") == proc_fd_match.group(0)
|
|
)
|
|
|
|
secret_rows = read_jsonl(zf, "rbac/secret_access.jsonl")
|
|
secret = next(
|
|
row
|
|
for row in secret_rows
|
|
if row.get("review_uid") == review_uid
|
|
and row.get("verb") == "get"
|
|
and row.get("resource") == "secrets"
|
|
and row.get("response_code") == 200
|
|
)
|
|
|
|
return {
|
|
"flow": "ad",
|
|
"review_short": review_short,
|
|
"fd_marker": runtime["marker"],
|
|
"secret_tail": secret["name"].rsplit("-", 1)[-1],
|
|
}
|
|
|
|
|
|
def solve_bundle(bundle_path: str | Path) -> str:
|
|
components = recover_components(bundle_path)
|
|
return "caplag{" + "_".join(
|
|
[
|
|
components["flow"],
|
|
components["review_short"],
|
|
components["fd_marker"],
|
|
components["secret_tail"],
|
|
]
|
|
) + "}"
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
if len(argv) != 2:
|
|
print(f"usage: {Path(argv[0]).name} public/admission_drift_case.zip", file=sys.stderr)
|
|
return 2
|
|
print(solve_bundle(argv[1]))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv))
|