edu/stem-launcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Render token scan as compact token table

Browse Source
This commit is contained in:
mpabi
2026-04-26 20:49:35 +02:00
parent a9d0d0117a
commit 2885d3a4fa
4 changed files with 399 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rvctl.py
+346 -165
View File
@@ -4,6 +4,7 @@
from __future__ import annotations
import argparse
import base64
import datetime as dt
import json
import os
@@ -13,6 +14,8 @@ import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from urllib import error as urlerror
from urllib import request as urlrequest
from urllib.parse import quote, unquote, urlsplit, urlunsplit
@@ -244,6 +247,17 @@ def server_type_from_split(config: WorkspaceConfig, split_url) -> str:
return "unknown"
def repo_parts_from_url(remote_url: str) -> tuple[str, str]:
split_url = urlsplit(remote_url)
repo_path = unquote(split_url.path).strip("/")
if repo_path.endswith(".git"):
repo_path = repo_path[:-4]
path_parts = [part for part in repo_path.split("/") if part]
if len(path_parts) < 2:
return "", ""
return path_parts[0], path_parts[1]
def server_info_from_url(config: WorkspaceConfig, remote_url: str) -> dict | None:
split_url = urlsplit(remote_url)
if split_url.scheme not in {"http", "https"}:
@@ -252,12 +266,15 @@ def server_info_from_url(config: WorkspaceConfig, remote_url: str) -> dict | Non
return None
endpoint, port_value = endpoint_from_split(split_url)
org_name, repo_name = repo_parts_from_url(remote_url)
return {
"endpoint": endpoint,
"type": server_type_from_split(config, split_url),
"scheme": split_url.scheme,
"host": split_url.hostname,
"port": port_value,
"org": org_name,
"repo": repo_name,
}
@@ -455,8 +472,11 @@ def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> dict:
"remote": remote_name,
"endpoint": "",
"type": "unsupported",
"org": "",
"repo": "",
"url_kind": "unsupported",
"user": "",
"token_value": "",
"result": "ignored",
}
)
@@ -471,8 +491,11 @@ def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> dict:
"remote": remote_name,
"endpoint": server_info["endpoint"],
"type": server_info["type"],
"org": server_info["org"],
"repo": server_info["repo"],
"url_kind": "plain",
"user": "",
"token_value": "",
"result": "no_credentials",
}
)
@@ -486,8 +509,11 @@ def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> dict:
"remote": remote_name,
"endpoint": server_info["endpoint"],
"type": server_info["type"],
"org": server_info["org"],
"repo": server_info["repo"],
"url_kind": "auth",
"user": user_name,
"token_value": token_value,
"result": "found",
}
)
@@ -575,6 +601,8 @@ def collect_repo_server_entries(config: WorkspaceConfig, repo_path: Path) -> tup
"auth_urls": 0,
"plain_urls": 0,
"users": {},
"tokens": [],
"plain_remotes": [],
},
)
entry["remotes"].add(remote_name)
@@ -586,10 +614,26 @@ def collect_repo_server_entries(config: WorkspaceConfig, repo_path: Path) -> tup
credentials = remote_credentials(remote_url)
if credentials is None:
entry["plain_urls"] += 1
entry["plain_remotes"].append(
{
"remote": remote_name,
"org": server_info["org"],
"repo": server_info["repo"],
}
)
continue
entry["auth_urls"] += 1
user_name, token_value = credentials
entry["tokens"].append(
{
"remote": remote_name,
"org": server_info["org"],
"repo": server_info["repo"],
"user": user_name,
"token_value": token_value,
}
)
user_entry = entry["users"].setdefault(user_name, {"tokens": {}})
tokens = user_entry.setdefault("tokens", {})
if token_value not in tokens.values():
@@ -657,7 +701,7 @@ def print_token_context(
print(f"token_path\t{config.token_path}")
print(f"repo_endpoints\t{len(repo_servers)}")
print(f"tokens_endpoints\t{len(store_servers)}")
print(f"items\t{endpoint_count}")
print(f"tokens\t{endpoint_count}")
def print_status_counts(status_counts: dict[str, int]) -> None:
@@ -667,19 +711,21 @@ def print_status_counts(status_counts: dict[str, int]) -> None:
print(f"{status_name}\t{status_counts[status_name]}")
TOKEN_ITEM_COLUMNS = [
("item", 4),
("source", 11),
("id", 10),
("server", 6),
("scheme", 6),
("host", 15),
("port", 5),
("credential", 10),
("repo_user", 9),
("store_user", 10),
("token", 5),
("sync", 4),
TOKEN_REF_WIDTH = 9
TOKEN_COLUMNS = [
("item", "item", 4),
("server", "server", 6),
("host", "host", 13),
("owner", "org", 9),
("repo_name", "repo", 11),
("user", "user", 4),
("remote", "remote", 6),
("token_ref", "token_ref", TOKEN_REF_WIDTH),
("token", "token", 12),
("scope_mask", "scope", 6),
("org_mask", "org", 6),
("repo_mask", "repo", 6),
]
@@ -691,24 +737,23 @@ def fit_cell(value: str, width: int) -> str:
return f"{value[:width - 3]}..."
def print_token_item_table(rows: list[dict[str, str]]) -> None:
print("items")
print(" ".join(fit_cell(name, width) for name, width in TOKEN_ITEM_COLUMNS).rstrip())
print(" ".join("-" * width for _, width in TOKEN_ITEM_COLUMNS).rstrip())
def print_fixed_table(title: str, columns: list[tuple[str, str, int]], rows: list[dict[str, str]]) -> None:
print(title)
print(" ".join(fit_cell(label, width) for _, label, width in columns).rstrip())
print(" ".join("-" * width for _, _, width in columns).rstrip())
for row in rows:
print(" ".join(fit_cell(row.get(name, ""), width) for name, width in TOKEN_ITEM_COLUMNS).rstrip())
print(" ".join(fit_cell(row.get(name, ""), width) for name, _, width in columns).rstrip())
def sync_mark(status_name: str, source: str) -> str:
if status_name == "diverged":
return "!"
if status_name in {"in_sync", "store_only", "store_ahead"}:
truth_source = "tokens.json"
elif status_name in {"repo_only", "repo_ahead"}:
truth_source = "repo"
else:
truth_source = ""
return "*" if source == truth_source else ""
def short_secret(secret_value: str) -> str:
if len(secret_value) <= 12:
return mask_secret(secret_value)
return f"{secret_value[:5]}...{secret_value[-4:]}"
def format_token_ref(token_name: str, marker: str) -> str:
name_width = TOKEN_REF_WIDTH - 2
return f"{fit_cell(token_name, name_width)}{marker:>2}"
def users_label(server_entry: dict | None) -> str:
@@ -741,6 +786,191 @@ def endpoint_column_values(endpoint: str, repo_entry: dict | None, store_entry:
return values
def token_store_rows(store_servers: dict[str, dict]) -> list[dict[str, str]]:
rows: list[dict[str, str]] = []
for endpoint, server_entry in sorted(store_servers.items()):
users = server_entry.get("users", {})
for user_name in sorted(users):
user_entry = users[user_name]
if not isinstance(user_entry, dict):
continue
for token_name, token_value in sorted(user_entry.get("tokens", {}).items()):
if isinstance(token_value, str):
rows.append(
{
"endpoint": endpoint,
"user": str(user_name),
"token_name": str(token_name),
"token_value": token_value,
}
)
return rows
def repo_token_rows(
endpoint_names: list[str],
repo_servers: dict[str, dict],
scan_rows: list[dict[str, str]] | None,
) -> list[dict[str, str]]:
if scan_rows is not None:
return [
{
"endpoint": row["endpoint"],
"remote": row["remote"],
"owner": row.get("org", ""),
"repo_name": row.get("repo", ""),
"user": row.get("user", ""),
"token_value": row.get("token_value", ""),
}
for row in scan_rows
if row.get("endpoint") and row.get("url_kind") == "auth" and row.get("token_value")
]
rows: list[dict[str, str]] = []
for endpoint in endpoint_names:
repo_entry = repo_servers.get(endpoint)
if repo_entry is None:
continue
for row in repo_entry.get("tokens", []):
rows.append(
{
"endpoint": endpoint,
"remote": row.get("remote", ""),
"owner": row.get("org", ""),
"repo_name": row.get("repo", ""),
"user": row.get("user", ""),
"token_value": row.get("token_value", ""),
}
)
return rows
def fallback_project_by_endpoint(
endpoint_names: list[str],
repo_servers: dict[str, dict],
scan_rows: list[dict[str, str]] | None,
) -> dict[str, dict[str, str]]:
fallback: dict[str, dict[str, str]] = {}
for row in scan_rows or []:
endpoint = row.get("endpoint", "")
if endpoint and endpoint not in fallback:
fallback[endpoint] = {
"remote": row.get("remote", ""),
"owner": row.get("org", ""),
"repo_name": row.get("repo", ""),
}
for endpoint in endpoint_names:
if endpoint in fallback:
continue
repo_entry = repo_servers.get(endpoint)
if repo_entry is None:
continue
for row in repo_entry.get("tokens", []) + repo_entry.get("plain_remotes", []):
fallback[endpoint] = {
"remote": row.get("remote", ""),
"owner": row.get("org", ""),
"repo_name": row.get("repo", ""),
}
break
return fallback
def api_get_json(api_url: str, token_value: str, user_name: str = "", basic: bool = False) -> object:
headers = {"Accept": "application/json"}
if basic:
raw_credentials = f"{user_name}:{token_value}".encode("utf-8")
headers["Authorization"] = "Basic " + base64.b64encode(raw_credentials).decode("ascii")
else:
headers["Authorization"] = f"token {token_value}"
request = urlrequest.Request(api_url, headers=headers)
with urlrequest.urlopen(request, timeout=5) as response:
return json.loads(response.read().decode("utf-8"))
def scope_mask(scopes: list[str]) -> str:
scope_set = {scope.lower() for scope in scopes}
return "".join(
[
"+" if "all" in scope_set else "-",
"+" if "write:repository" in scope_set else "-",
"+" if "read:repository" in scope_set else "-",
"+" if {"read:organization", "write:organization"} & scope_set else "-",
"+" if "public-only" in scope_set else "-",
]
)
def org_permission_mask(org_permissions: dict) -> str:
return "".join(
[
"+" if org_permissions.get("is_owner") else "-",
"+" if org_permissions.get("is_admin") else "-",
"+" if org_permissions.get("can_write") else "-",
"+" if org_permissions.get("can_read") else "-",
"+" if org_permissions.get("can_create_repository") else "-",
]
)
def repo_permission_mask(repo_permissions: dict) -> str:
permission = str(repo_permissions.get("permission", "none")).lower()
levels = {
"owner": "++++",
"admin": "-+++",
"write": "--++",
"read": "---+",
}
return levels.get(permission, "----")
def load_gitea_authz(endpoint: str, user_name: str, token_name: str, token_value: str, owner: str, repo_name: str) -> dict:
api_base = endpoint.rstrip("/") + "/api/v1"
result = {
"ok": False,
"scope_mask": "!!!!!",
"org_mask": "!!!!!",
"repo_mask": "!!!!",
}
try:
tokens_url = f"{api_base}/users/{quote(user_name, safe='')}/tokens"
token_list = api_get_json(tokens_url, token_value, user_name=user_name, basic=True)
token_last8 = token_value[-8:]
token_entry = next(
(
entry
for entry in token_list
if isinstance(entry, dict) and entry.get("token_last_eight") == token_last8
),
None,
)
scopes = token_entry.get("scopes", []) if isinstance(token_entry, dict) else []
if not isinstance(scopes, list):
scopes = []
result["scope_mask"] = scope_mask([str(scope) for scope in scopes])
if owner:
org_url = f"{api_base}/users/{quote(user_name, safe='')}/orgs/{quote(owner, safe='')}/permissions"
org_permissions = api_get_json(org_url, token_value)
if isinstance(org_permissions, dict):
result["org_mask"] = org_permission_mask(org_permissions)
if owner and repo_name:
repo_url = (
f"{api_base}/repos/{quote(owner, safe='')}/{quote(repo_name, safe='')}"
f"/collaborators/{quote(user_name, safe='')}/permission"
)
repo_permissions = api_get_json(repo_url, token_value)
if isinstance(repo_permissions, dict):
result["repo_mask"] = repo_permission_mask(repo_permissions)
result["ok"] = True
return result
except (OSError, ValueError, urlerror.URLError, urlerror.HTTPError):
return result
def print_token_item_rows(
endpoint_names: list[str],
repo_servers: dict[str, dict],
@@ -749,132 +979,100 @@ def print_token_item_rows(
) -> dict[str, int]:
status_names = ["in_sync", "repo_only", "store_only", "repo_ahead", "store_ahead", "diverged"]
status_counts = {status_name: 0 for status_name in status_names}
scan_rows_by_endpoint: dict[str, list[dict[str, str]]] = {}
for row in scan_rows or []:
endpoint = row.get("endpoint", "")
if endpoint:
scan_rows_by_endpoint.setdefault(endpoint, []).append(row)
item_rows: list[dict[str, str]] = []
for index, endpoint in enumerate(endpoint_names, start=1):
item_id = str(index)
for endpoint in endpoint_names:
repo_entry = repo_servers.get(endpoint)
store_entry = store_servers.get(endpoint)
status_name, _, _ = compare_server_entries(repo_entry, store_entry)
if status_name in status_counts:
status_counts[status_name] += 1
endpoint_values = endpoint_column_values(endpoint, repo_entry, store_entry)
repo_users_label = users_label(repo_entry)
store_users_label = users_label(store_entry)
repo_rows = scan_rows_by_endpoint.get(endpoint, [])
if repo_rows:
for row in repo_rows:
token_label = "remote" if row["url_kind"] == "auth" else ""
item_rows.append(
{
"item": item_id,
"source": "repo",
"id": row["remote"],
**endpoint_values,
"credential": row["url_kind"],
"repo_user": row["user"],
"store_user": store_users_label,
"token": "yes" if token_label else "no",
"sync": sync_mark(status_name, "repo"),
}
)
elif repo_entry is None:
item_rows.append(
{
"item": item_id,
"source": "repo",
**endpoint_values,
"credential": "missing",
"store_user": store_users_label,
"token": "no",
"sync": sync_mark(status_name, "repo"),
}
)
else:
repo_kind, repo_result = repo_result_for_entry(repo_entry)
remote_names = ",".join(sorted(repo_entry.get("remotes", set())))
repo_users = token_name_rows(repo_entry.get("users", {}))
if repo_users:
for user_name, _ in repo_users:
token_label = "remote" if repo_kind == "auth" else ""
item_rows.append(
{
"item": item_id,
"source": "repo",
"id": remote_names,
**endpoint_values,
"credential": repo_kind,
"repo_user": user_name,
"store_user": store_users_label,
"token": "yes" if token_label else "no",
"sync": sync_mark(status_name, "repo"),
}
)
else:
item_rows.append(
{
"item": item_id,
"source": "repo",
"id": remote_names,
**endpoint_values,
"credential": repo_kind,
"store_user": store_users_label,
"token": "yes" if repo_result == "present" else "no",
"sync": sync_mark(status_name, "repo"),
}
)
store_rows = token_store_rows(store_servers)
store_by_value: dict[tuple[str, str, str], dict[str, str]] = {}
for row in store_rows:
store_by_value.setdefault((row["endpoint"], row["user"], row["token_value"]), row)
if store_entry is None:
item_rows.append(
{
"item": item_id,
"source": "tokens.json",
**endpoint_values,
"credential": "missing",
"repo_user": repo_users_label,
"token": "no",
"sync": sync_mark(status_name, "tokens.json"),
}
)
endpoint_values_by_name = {
endpoint: endpoint_column_values(endpoint, repo_servers.get(endpoint), store_servers.get(endpoint))
for endpoint in endpoint_names
}
project_fallback = fallback_project_by_endpoint(endpoint_names, repo_servers, scan_rows)
authz_cache: dict[tuple[str, str, str, str, str], dict] = {}
used_store_keys: set[tuple[str, str, str, str]] = set()
token_rows: list[dict[str, str]] = []
def append_token_row(
endpoint: str,
user_name: str,
remote_name: str,
token_name: str,
token_value: str,
marker: str,
owner: str,
repo_name: str,
authz: dict | None = None,
) -> None:
endpoint_values = endpoint_values_by_name.get(endpoint, endpoint_column_values(endpoint, None, None))
token_rows.append(
{
"item": str(len(token_rows) + 1),
"server": endpoint_values["server"],
"host": endpoint_values["host"],
"owner": owner,
"repo_name": repo_name,
"user": user_name,
"remote": remote_name,
"token_ref": format_token_ref(token_name, marker),
"token": short_secret(token_value),
"scope_mask": authz.get("scope_mask", "?????") if authz else "?????",
"org_mask": authz.get("org_mask", "?????") if authz else "?????",
"repo_mask": authz.get("repo_mask", "????") if authz else "????",
}
)
for row in repo_token_rows(endpoint_names, repo_servers, scan_rows):
endpoint = row["endpoint"]
token_value = row["token_value"]
store_row = store_by_value.get((endpoint, row["user"], token_value))
token_name = store_row["token_name"] if store_row else ""
marker = "R"
authz = None
if store_row:
used_store_keys.add((endpoint, store_row["user"], store_row["token_name"], token_value))
authz_key = (endpoint, row["user"], token_value, row["owner"], row["repo_name"])
authz = authz_cache.get(authz_key)
if authz is None:
authz = load_gitea_authz(endpoint, row["user"], token_name, token_value, row["owner"], row["repo_name"])
authz_cache[authz_key] = authz
marker = "*" if authz.get("ok") else "!"
append_token_row(
endpoint,
row["user"],
row["remote"],
token_name,
token_value,
marker,
row["owner"],
row["repo_name"],
authz if marker == "*" else None,
)
for row in store_rows:
store_key = (row["endpoint"], row["user"], row["token_name"], row["token_value"])
if store_key in used_store_keys:
continue
fallback = project_fallback.get(row["endpoint"], {})
append_token_row(
row["endpoint"],
row["user"],
"",
row["token_name"],
row["token_value"],
"S",
fallback.get("owner", ""),
fallback.get("repo_name", ""),
)
store_users = token_name_rows(store_entry.get("users", {}))
if not store_users:
item_rows.append(
{
"item": item_id,
"source": "tokens.json",
**endpoint_values,
"credential": "store",
"repo_user": repo_users_label,
"token": "no",
"sync": sync_mark(status_name, "tokens.json"),
}
)
continue
for user_name, token_name in store_users:
item_rows.append(
{
"item": item_id,
"source": "tokens.json",
"credential": "store",
"id": token_name,
**endpoint_values,
"repo_user": repo_users_label,
"store_user": user_name,
"token": "yes" if token_name else "no",
"sync": sync_mark(status_name, "tokens.json"),
}
)
print_token_item_table(item_rows)
print_fixed_table("tokens", TOKEN_COLUMNS, token_rows)
return status_counts
@@ -972,22 +1170,6 @@ def run_tokens_scan(config: WorkspaceConfig, args: argparse.Namespace) -> None:
token_data = load_token_store(config)
store_servers = token_data.get("servers", {})
endpoint_names = sorted(set(repo_servers) | set(store_servers))
if args.verbose:
print_token_context(repo_root, config, repo_servers, store_servers, len(endpoint_names))
print()
print("scan")
print("item\tvalue")
print(f"remotes\t{report['remotes']}")
print(f"urls\t{report['urls']}")
print(f"scanned\t{report['scanned']}")
print(f"auth_urls\t{report['auth_urls']}")
print(f"plain_urls\t{report['plain_urls']}")
print(f"unsupported_urls\t{report['unsupported_urls']}")
print(f"found\t{report['found']}")
print(f"added\t{report['added']}")
print(f"servers\t{report['servers']}")
print()
print_token_item_rows(endpoint_names, repo_servers, store_servers, report["remote_rows"])
@@ -1386,7 +1568,7 @@ def print_main_overview(config_path: Path) -> None:
print_table(
["step", "command", "result"],
[
["1", "./rvctl tokens scan", "pair repo remotes with tokens.json entries"],
["1", "./rvctl tokens scan", "print token table from repo remotes and tokens.json"],
["2", "./rvctl list-series", "choose a series"],
["3", "./rvctl list-cards inf", "choose a card"],
["4", "./rvctl tmux-container inf <card>", "start the working container"],
@@ -1413,7 +1595,7 @@ def print_tokens_overview() -> None:
print_table(
["command", "common options", "direction", "purpose"],
[
["scan", "[--repo PATH]", "repo + tokens.json", "pair endpoints and show where token refs are stored"],
["scan", "[--repo PATH]", "repo + tokens.json", "print token table with remote/store marker and auth masks"],
["read", "[--server ENDPOINT]", "tokens.json", "show servers, users and token names"],
["stats", "[--repo PATH]", "repo + tokens.json", "compare remote URLs with local token store"],
["write", "--remote R [--replace]", "tokens.json -> repo", "write selected token into a remote URL"],
@@ -1524,7 +1706,6 @@ def build_parser() -> argparse.ArgumentParser:
help="Scan repo remotes and tokens.json, then print paired endpoint rows.",
)
add_repo_option(tokens_scan_parser)
tokens_scan_parser.add_argument("--verbose", action="store_true", help="Print context and scan counters before item rows.")
tokens_read_parser = tokens_subparsers.add_parser(
"read",