#!/usr/bin/env python3 """rvctl command-line implementation.""" from __future__ import annotations import argparse import base64 import datetime as dt import json import os import re import shlex 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 def expand_path(raw_path: str, base_dir: Path) -> Path: expanded = os.path.expandvars(os.path.expanduser(raw_path)) path = Path(expanded) if not path.is_absolute(): path = base_dir / path return path.resolve() @dataclass class GitConfig: base_url: str source_org: str answer_org: str source_remote: str answer_remote: str origin_remote: str fallback_branch: str @dataclass class WorkspaceConfig: config_path: Path original_root: Path original_series_root: Path workspace_root: Path series_root: Path socket_root: Path token_path: Path tools_root_candidates: list[Path] tools_root: Path defaults: dict[str, str] git: GitConfig def load_config(config_path: Path) -> WorkspaceConfig: raw = json.loads(config_path.read_text(encoding="utf-8")) base_dir = config_path.parent original_root = expand_path(raw.get("original_root", "~/dev/edu/repos/rv"), base_dir) workspace_root = expand_path(raw.get("workspace_root", "~/dev/workspace/rv"), base_dir) original_series_root = expand_path(raw.get("original_series_root", str(original_root / "series")), base_dir) series_root = expand_path(raw.get("series_root", str(workspace_root / "series")), base_dir) socket_root = expand_path(raw.get("socket_root", str(workspace_root / "sockets")), base_dir) token_path = expand_path(raw.get("token_file", str(workspace_root / "tokens" / "tokens.json")), base_dir) candidate_values = raw.get("tools_root_candidates") if not candidate_values: candidate_values = [ raw.get("tools_root", str(workspace_root / "tools" / "rv32i-hazard3-env")), str(original_root / "rv32i-hazard3-env"), ] tools_root_candidates = [expand_path(value, base_dir) for value in candidate_values] tools_root = next((path for path in tools_root_candidates if path.exists()), tools_root_candidates[0]) defaults = raw.get("defaults", {}) git_raw = raw.get("git", {}) git = GitConfig( base_url=git_raw.get("base_url", "http://77.90.8.171:3001").rstrip("/"), source_org=git_raw.get("source_org", "edu-inf"), answer_org=git_raw.get("answer_org", "zsl-inf"), source_remote=git_raw.get("source_remote", "r1"), answer_remote=git_raw.get("answer_remote", "a1"), origin_remote=git_raw.get("origin_remote", "origin"), fallback_branch=git_raw.get("fallback_branch", "build"), ) return WorkspaceConfig( config_path=config_path, original_root=original_root, original_series_root=original_series_root, workspace_root=workspace_root, series_root=series_root, socket_root=socket_root, token_path=token_path, tools_root_candidates=tools_root_candidates, tools_root=tools_root, defaults=defaults, git=git, ) def series_directories(config: WorkspaceConfig) -> list[Path]: if not config.series_root.is_dir(): raise SystemExit(f"Missing series root: {config.series_root}") return sorted(path for path in config.series_root.iterdir() if path.is_dir()) def card_directories(series_path: Path) -> list[Path]: return sorted(path for path in series_path.iterdir() if path.is_dir()) def read_card_title(card_path: Path) -> str: readme_path = card_path / "README.md" if not readme_path.is_file(): return "" for line in readme_path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if stripped.startswith("#"): return stripped.lstrip("#").strip() return "" def resolve_selector( config: WorkspaceConfig, series_arg: str | None, card_arg: str | None, ) -> tuple[str, str]: default_series = config.defaults.get("series") default_card = config.defaults.get("card") if series_arg and "/" in series_arg: if card_arg: raise SystemExit("Pass either ' ' or '', not both forms at once.") series_value, card_value = series_arg.split("/", 1) return series_value, card_value if series_arg and not card_arg: if (config.series_root / series_arg).is_dir(): if not default_card: raise SystemExit("Missing default card in config.") return series_arg, default_card if default_series and (config.series_root / default_series / series_arg).is_dir(): return default_series, series_arg series_value = series_arg or default_series card_value = card_arg or default_card if not series_value or not card_value: raise SystemExit("Missing series/card selector and no defaults are configured.") return series_value, card_value def resolve_card_path(config: WorkspaceConfig, series_name: str, card_name: str) -> Path: card_path = config.series_root / series_name / card_name if not card_path.is_dir(): raise SystemExit(f"Missing card path: {card_path}") return card_path def print_series(config: WorkspaceConfig) -> None: for series_path in series_directories(config): card_count = len(card_directories(series_path)) print(f"{series_path.name}\t{card_count}\t{series_path}") def print_cards(config: WorkspaceConfig, series_arg: str | None) -> None: series_name = series_arg or config.defaults.get("series") if not series_name: raise SystemExit("Missing series name and no default series is configured.") series_path = config.series_root / series_name if not series_path.is_dir(): raise SystemExit(f"Missing series path: {series_path}") for card_path in card_directories(series_path): title = read_card_title(card_path) if title: print(f"{card_path.name}\t{title}\t{card_path}") else: print(f"{card_path.name}\t{card_path}") def slug_value(raw_value: str, field_name: str) -> str: slug = re.sub(r"[^a-z0-9._-]+", "-", raw_value.strip().lower()).strip("-") if not slug: raise SystemExit(f"Empty {field_name} after slug conversion: {raw_value!r}") return slug def git_capture(repo_path: Path, git_args: list[str]) -> str | None: result = subprocess.run( ["git", "-C", str(repo_path)] + git_args, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=False, ) if result.returncode != 0: return None return result.stdout.strip() def git_repo_root(repo_path: Path) -> Path | None: repo_root = git_capture(repo_path, ["rev-parse", "--show-toplevel"]) if not repo_root: return None return Path(repo_root) def default_port_for_scheme(scheme: str) -> int | None: if scheme == "http": return 80 if scheme == "https": return 443 return None def endpoint_from_split(split_url) -> tuple[str, int | None]: host_name = split_url.hostname if not host_name: raise SystemExit(f"Missing hostname in URL: {split_url.geturl()}") port_value = split_url.port or default_port_for_scheme(split_url.scheme) default_port = default_port_for_scheme(split_url.scheme) if port_value is not None and port_value != default_port: endpoint = f"{split_url.scheme}://{host_name}:{port_value}" else: endpoint = f"{split_url.scheme}://{host_name}" return endpoint, port_value def server_type_from_split(config: WorkspaceConfig, split_url) -> str: endpoint, _ = endpoint_from_split(split_url) base_split = urlsplit(config.git.base_url) base_endpoint, _ = endpoint_from_split(base_split) if endpoint == base_endpoint: return "gitea" host_name = (split_url.hostname or "").lower() if host_name == "github.com" or host_name.endswith(".github.com"): return "github" if "gitlab" in host_name: return "gitlab" if "gitea" in host_name: return "gitea" 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"}: return None if split_url.hostname is None: 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, } def empty_token_store() -> dict: return {"version": 3, "tokens": []} def next_token_name(tokens: dict[str, str]) -> str: index = 1 while f"t{index}" in tokens: index += 1 return f"t{index}" def server_record_from_info(server_info: dict) -> dict: return { "type": str(server_info.get("type", "unknown")), "endpoint": str(server_info.get("endpoint", "")), "scheme": str(server_info.get("scheme", "")), "host": str(server_info.get("host", "")), "port": server_info.get("port"), } def normalize_server_record(config: WorkspaceConfig, raw_server: dict, endpoint: str | None = None) -> dict | None: if not isinstance(raw_server, dict): return None endpoint_value = str(raw_server.get("endpoint") or endpoint or "") if not endpoint_value: return None split_endpoint = urlsplit(endpoint_value) host_value = raw_server.get("host") or split_endpoint.hostname or "" scheme_value = raw_server.get("scheme") or split_endpoint.scheme port_value = raw_server.get("port") if port_value is None and split_endpoint.port is not None: port_value = split_endpoint.port server_type = raw_server.get("type") or "unknown" if server_type == "unknown" and scheme_value and host_value: server_url = endpoint_value server_type = server_type_from_split(config, urlsplit(server_url)) return { "type": str(server_type), "endpoint": endpoint_value, "scheme": str(scheme_value), "host": str(host_value), "port": port_value, } def normalize_token_entry(raw_token_entry) -> str | dict | None: if isinstance(raw_token_entry, str): return raw_token_entry if raw_token_entry else None if not isinstance(raw_token_entry, dict): return None token_value = raw_token_entry.get("value") if not isinstance(token_value, str) or not token_value: return None token_entry = {"value": token_value} expires_at = raw_token_entry.get("expires_at") if isinstance(expires_at, str) and expires_at: token_entry["expires_at"] = expires_at for field_name in ["remote", "org", "repo"]: field_value = raw_token_entry.get(field_name) if isinstance(field_value, str) and field_value: token_entry[field_name] = field_value return token_entry def token_entry_value(token_entry) -> str: if isinstance(token_entry, str): return token_entry if isinstance(token_entry, dict) and isinstance(token_entry.get("value"), str): return token_entry["value"] return "" def token_entry_expires_at(token_entry) -> str: if isinstance(token_entry, dict) and isinstance(token_entry.get("expires_at"), str): return token_entry["expires_at"] return "" def token_entry_field(token_entry, field_name: str) -> str: if isinstance(token_entry, dict) and isinstance(token_entry.get(field_name), str): return token_entry[field_name] return "" def token_value_exists(tokens: dict, token_value: str) -> bool: return any(token_entry_value(existing_token) == token_value for existing_token in tokens.values()) def normalize_remote_record(raw_remote: dict) -> dict | None: if not isinstance(raw_remote, dict): return None remote_name = raw_remote.get("name") or raw_remote.get("remote") if not isinstance(remote_name, str) or not remote_name: return None remote_record = {"name": remote_name, "org": "", "repo": ""} for field_name in ["org", "repo", "org_perm", "repo_perm"]: field_value = raw_remote.get(field_name) if isinstance(field_value, str) and field_value: remote_record[field_name] = field_value return remote_record def valid_mask(raw_value: str, allowed: str, width: int) -> bool: return len(raw_value) == width and all(char in allowed for char in raw_value) def normalize_v3_token(config: WorkspaceConfig, raw_token: dict) -> dict | None: if not isinstance(raw_token, dict): return None token_id = raw_token.get("token_id") or raw_token.get("id") or raw_token.get("token_ref") token_value = raw_token.get("value") if token_value is None: token_value = raw_token.get("token") if not isinstance(token_id, str) or not token_id: return None if not isinstance(token_value, str): return None server_record = normalize_server_record(config, raw_token.get("server", {})) if server_record is None: return None token_record = { "token_id": token_id, "value": token_value, "server": server_record, "remotes": [], } for field_name in ["user", "valid", "expires_at"]: field_value = raw_token.get(field_name) if isinstance(field_value, str) and field_value: token_record[field_name] = field_value scope_value = raw_token.get("scope") if isinstance(scope_value, str) and valid_mask(scope_value, "rw?!-", 9): token_record["scope"] = scope_value raw_remotes = raw_token.get("remotes", []) if isinstance(raw_remotes, dict): raw_remotes = [ {"name": remote_name, **remote_value} for remote_name, remote_value in raw_remotes.items() if isinstance(remote_value, dict) ] if isinstance(raw_remotes, list): for raw_remote in raw_remotes: remote_record = normalize_remote_record(raw_remote) if remote_record is not None: token_record["remotes"].append(remote_record) return token_record def append_token_record(token_data: dict, token_record: dict) -> None: tokens = token_data.setdefault("tokens", []) token_key = ( token_record.get("server", {}).get("endpoint", ""), token_record.get("token_id", ""), token_record.get("value", ""), ) for existing_record in tokens: existing_key = ( existing_record.get("server", {}).get("endpoint", ""), existing_record.get("token_id", ""), existing_record.get("value", ""), ) if existing_key == token_key: existing_remotes = existing_record.setdefault("remotes", []) existing_remote_names = {remote.get("name", "") for remote in existing_remotes} for remote_record in token_record.get("remotes", []): if remote_record.get("name", "") not in existing_remote_names: existing_remotes.append(remote_record) return tokens.append(token_record) def parse_datetime(raw_value: str) -> dt.datetime | None: if not raw_value: return None try: normalized_value = raw_value.replace("Z", "+00:00") return dt.datetime.fromisoformat(normalized_value) except ValueError: return None def is_expired(raw_value: str) -> bool: expires_at = parse_datetime(raw_value) if expires_at is None: return False if expires_at.tzinfo is None: now = dt.datetime.now() else: now = dt.datetime.now(expires_at.tzinfo) return expires_at <= now def valid_label(api_accepts_token: bool, expires_at: str) -> str: if not api_accepts_token or (expires_at and is_expired(expires_at)): return "invalid" if expires_at: return expires_at return "forever" def copy_legacy_users(token_data: dict, server_info: dict, raw_users: dict) -> dict: for user_name, user_entry in raw_users.items(): if not isinstance(user_entry, dict): continue raw_tokens = user_entry.get("tokens", {}) if not isinstance(raw_tokens, dict): continue for token_name, raw_token_entry in raw_tokens.items(): token_entry = normalize_token_entry(raw_token_entry) if token_entry is None: continue token_value = token_entry_value(token_entry) token_record = { "token_id": str(token_name), "value": token_value, "server": server_record_from_info(server_info), "user": str(user_name), "remotes": [], } expires_at = token_entry_expires_at(token_entry) if expires_at: token_record["expires_at"] = expires_at remote_name = token_entry_field(token_entry, "remote") if remote_name: remote_record = {"name": remote_name, "org": "", "repo": ""} for source_name, target_name in [("org", "org"), ("repo", "repo")]: source_value = token_entry_field(token_entry, source_name) if source_value: remote_record[target_name] = source_value token_record["remotes"].append(remote_record) append_token_record(token_data, token_record) return token_data def normalize_token_store(config: WorkspaceConfig, raw_data: dict) -> dict: if raw_data.get("version") == 3 and isinstance(raw_data.get("tokens"), list): token_data = empty_token_store() for raw_token in raw_data["tokens"]: token_record = normalize_v3_token(config, raw_token) if token_record is not None: append_token_record(token_data, token_record) return token_data if "servers" in raw_data and isinstance(raw_data["servers"], dict): token_data = empty_token_store() for endpoint, server_entry in raw_data["servers"].items(): if not isinstance(server_entry, dict): continue base_server_info = { "endpoint": str(endpoint), "type": server_entry.get("type", "unknown"), "scheme": server_entry.get("scheme", ""), "host": server_entry.get("host", ""), "port": server_entry.get("port"), } copy_legacy_users(token_data, base_server_info, server_entry.get("users", {})) return token_data token_data = empty_token_store() if "users" in raw_data and isinstance(raw_data["users"], dict): base_server_info = server_info_from_url(config, config.git.base_url) if base_server_info is None: raise SystemExit(f"Invalid git base URL: {config.git.base_url}") copy_legacy_users(token_data, base_server_info, raw_data["users"]) return token_data def load_token_store(config: WorkspaceConfig, write_normalized: bool = True) -> dict: if not config.token_path.exists(): return empty_token_store() raw_data = json.loads(config.token_path.read_text(encoding="utf-8")) if not isinstance(raw_data, dict): raise SystemExit(f"Invalid token store format: {config.token_path}") normalized_data = normalize_token_store(config, raw_data) if write_normalized and raw_data != normalized_data: write_token_store(config, normalized_data) return normalized_data def write_token_store(config: WorkspaceConfig, token_data: dict) -> None: config.token_path.parent.mkdir(parents=True, exist_ok=True) os.chmod(config.token_path.parent, 0o700) temp_path = config.token_path.with_suffix(config.token_path.suffix + ".tmp") temp_path.write_text(json.dumps(token_data, indent=2, sort_keys=True) + "\n", encoding="utf-8") os.chmod(temp_path, 0o600) temp_path.replace(config.token_path) os.chmod(config.token_path, 0o600) def token_server_endpoint(token_record: dict) -> str: server_record = token_record.get("server", {}) if not isinstance(server_record, dict): return "" return str(server_record.get("endpoint", "")) def find_token_record(token_data: dict, endpoint: str, token_id: str) -> dict | None: for token_record in token_data.get("tokens", []): if token_server_endpoint(token_record) == endpoint and token_record.get("token_id") == token_id: return token_record return None def find_orphan_token_by_value(token_data: dict, endpoint: str, token_value: str) -> dict | None: for token_record in token_data.get("tokens", []): if token_server_endpoint(token_record) != endpoint: continue if token_record.get("value") != token_value: continue if token_record.get("remotes"): continue return token_record return None def find_remote_record(token_record: dict, remote_name: str) -> dict | None: for remote_record in token_record.setdefault("remotes", []): if remote_record.get("name") == remote_name: return remote_record return None def upsert_remote_record(token_record: dict, remote_name: str, server_info: dict) -> tuple[dict, bool]: remote_record = find_remote_record(token_record, remote_name) changed = False if remote_record is None: remote_record = {"name": remote_name, "org": "", "repo": ""} token_record.setdefault("remotes", []).append(remote_record) changed = True for source_name, target_name in [("org", "org"), ("repo", "repo")]: source_value = server_info.get(source_name) if isinstance(source_value, str) and source_value and remote_record.get(target_name) != source_value: remote_record[target_name] = source_value changed = True return remote_record, changed def apply_authz_to_token(token_record: dict, remote_record: dict, authz: dict) -> bool: changed = False for source_name, target_name in [ ("user", "user"), ("valid", "valid"), ("scope_mask", "scope"), ]: source_value = authz.get(source_name) if isinstance(source_value, str) and source_value and token_record.get(target_name) != source_value: token_record[target_name] = source_value changed = True for source_name, target_name in [ ("org_mask", "org_perm"), ("repo_mask", "repo_perm"), ]: source_value = authz.get(source_name) if isinstance(source_value, str) and source_value and remote_record.get(target_name) != source_value: remote_record[target_name] = source_value changed = True return changed def register_token(token_data: dict, server_info: dict, remote_name: str, token_id: str, token_value: str) -> str: endpoint = server_info["endpoint"] token_record = find_token_record(token_data, endpoint, token_id) if token_record is None: token_record = find_orphan_token_by_value(token_data, endpoint, token_value) if token_record is None: token_record = { "token_id": token_id, "value": token_value, "server": server_record_from_info(server_info), "remotes": [], } token_data.setdefault("tokens", []).append(token_record) result = "added" else: token_record["token_id"] = token_id result = "updated" else: result = "existing" if token_record.get("value") != token_value: token_record["value"] = token_value result = "updated" if token_record.get("server") != server_record_from_info(server_info): token_record["server"] = server_record_from_info(server_info) result = "updated" remote_record, remote_changed = upsert_remote_record(token_record, remote_name, server_info) if remote_changed and result == "existing": result = "updated" authz = load_gitea_authz( endpoint, token_id, token_value, token_record.get("expires_at", ""), remote_record.get("org", ""), remote_record.get("repo", ""), ) if apply_authz_to_token(token_record, remote_record, authz) and result == "existing": result = "updated" return result def remote_urls(repo_path: Path) -> list[str]: return [remote_url for _, remote_url in remote_name_urls(repo_path)] def remote_name_urls(repo_path: Path) -> list[tuple[str, str]]: remote_output = git_capture(repo_path, ["remote"]) if not remote_output: return [] urls: list[tuple[str, str]] = [] seen: set[tuple[str, str]] = set() for remote_name in remote_output.splitlines(): url_output = git_capture(repo_path, ["remote", "get-url", "--all", remote_name]) if not url_output: continue for remote_url in url_output.splitlines(): key = (remote_name, remote_url) if key in seen: continue seen.add(key) urls.append(key) return urls def remote_credentials(remote_url: str) -> tuple[str, str] | None: split_url = urlsplit(remote_url) if split_url.scheme not in {"http", "https"}: return None if split_url.username is None or split_url.password is None: return None return unquote(split_url.username), unquote(split_url.password) def scan_repo_remotes(config: WorkspaceConfig, repo_path: Path) -> dict: repo_root = git_repo_root(repo_path) if repo_root is None: return { "repo_root": None, "remotes": 0, "urls": 0, "scanned": 0, "auth_urls": 0, "plain_urls": 0, "unsupported_urls": 0, "found": 0, "added": 0, "updated": 0, "servers": 0, "remote_rows": [], "found_credentials": [], "changed": False, } found_credentials: list[tuple[int, dict, str, str, str]] = [] remote_entries = remote_name_urls(repo_root) remote_names = {remote_name for remote_name, _ in remote_entries} touched_servers: set[str] = set() remote_rows: list[dict[str, str]] = [] auth_urls = 0 plain_urls = 0 unsupported_urls = 0 for remote_name, remote_url in remote_entries: server_info = server_info_from_url(config, remote_url) if server_info is None: unsupported_urls += 1 remote_rows.append( { "remote": remote_name, "endpoint": "", "type": "unsupported", "org": "", "repo": "", "url_kind": "unsupported", "token_id": "", "token_value": "", "result": "ignored", } ) continue touched_servers.add(server_info["endpoint"]) credentials = remote_credentials(remote_url) if credentials is None: plain_urls += 1 remote_rows.append( { "remote": remote_name, "endpoint": server_info["endpoint"], "type": server_info["type"], "org": server_info["org"], "repo": server_info["repo"], "url_kind": "plain", "token_id": "", "token_value": "", "result": "no_credentials", } ) continue auth_urls += 1 token_id, token_value = credentials row_index = len(remote_rows) remote_rows.append( { "remote": remote_name, "endpoint": server_info["endpoint"], "type": server_info["type"], "org": server_info["org"], "repo": server_info["repo"], "url_kind": "auth", "token_id": token_id, "token_value": token_value, "result": "found", } ) found_credentials.append((row_index, server_info, remote_name, token_id, token_value)) return { "repo_root": repo_root, "remotes": len(remote_names), "urls": len(remote_entries), "scanned": auth_urls + plain_urls, "auth_urls": auth_urls, "plain_urls": plain_urls, "unsupported_urls": unsupported_urls, "found": len(found_credentials), "added": 0, "updated": 0, "servers": len(touched_servers), "remote_rows": remote_rows, "found_credentials": found_credentials, "changed": False, } def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> dict: report = scan_repo_remotes(config, repo_path) found_credentials = report.get("found_credentials", []) if not found_credentials: return report token_data = load_token_store(config) added = 0 updated = 0 for row_index, server_info, remote_name, token_id, token_value in found_credentials: result = register_token(token_data, server_info, remote_name, token_id, token_value) if result == "added": added += 1 elif result == "updated": updated += 1 report["remote_rows"][row_index]["result"] = result changed = added > 0 or updated > 0 if changed or not config.token_path.exists(): write_token_store(config, token_data) report["added"] = added report["updated"] = updated report["changed"] = changed return report def count_server_tokens(server_entry: dict) -> int: return len(server_entry.get("tokens", [])) def mask_secret(secret_value: str) -> str: if len(secret_value) <= 4: return "*" * len(secret_value) return f"{secret_value[:2]}{'*' * (len(secret_value) - 4)}{secret_value[-2:]}" def store_servers_from_tokens(token_data: dict) -> dict[str, dict]: servers: dict[str, dict] = {} for token_record in token_data.get("tokens", []): server_record = token_record.get("server", {}) if not isinstance(server_record, dict): continue endpoint = str(server_record.get("endpoint", "")) if not endpoint: continue server_entry = servers.setdefault( endpoint, { "type": server_record.get("type", "unknown"), "scheme": server_record.get("scheme", ""), "host": server_record.get("host", ""), "port": server_record.get("port"), "tokens": [], }, ) for field_name in ["type", "scheme", "host", "port"]: if server_entry.get(field_name) in {"", None, "unknown"} and server_record.get(field_name) not in {"", None}: server_entry[field_name] = server_record.get(field_name) remotes = token_record.get("remotes", []) if not isinstance(remotes, list) or not remotes: remotes = [{}] for remote_record in remotes: if not isinstance(remote_record, dict): continue server_entry["tokens"].append( { "endpoint": endpoint, "token_id": str(token_record.get("token_id", "")), "token_value": str(token_record.get("value", "")), "user": str(token_record.get("user", "")), "valid": str(token_record.get("valid", "")), "scope_mask": str(token_record.get("scope", "")), "expires_at": str(token_record.get("expires_at", "")), "remote": str(remote_record.get("name", "")), "owner": str(remote_record.get("org", "")), "repo_name": str(remote_record.get("repo", "")), "org_mask": str(remote_record.get("org_perm", "")), "repo_mask": str(remote_record.get("repo_perm", "")), } ) return servers def token_store_servers(token_data: dict) -> list[tuple[str, dict]]: return sorted(store_servers_from_tokens(token_data).items()) def token_pairs_from_entry(server_entry: dict) -> set[tuple[str, str, str, str, str]]: token_pairs: set[tuple[str, str, str, str, str]] = set() for row in server_entry.get("tokens", []): token_id = row.get("token_id", "") token_value = row.get("token_value", "") if token_id and token_value: token_pairs.add( ( str(row.get("remote", "")), str(token_id), str(token_value), str(row.get("org", row.get("owner", ""))), str(row.get("repo", row.get("repo_name", ""))), ) ) return token_pairs def count_users(users_data: dict) -> int: return len(users_data) def collect_repo_server_entries(config: WorkspaceConfig, repo_path: Path) -> tuple[Path | None, dict[str, dict]]: repo_root = git_repo_root(repo_path) if repo_root is None: return None, {} repo_servers: dict[str, dict] = {} for remote_name, remote_url in remote_name_urls(repo_root): server_info = server_info_from_url(config, remote_url) if server_info is None: continue endpoint = server_info["endpoint"] entry = repo_servers.setdefault( endpoint, { "type": server_info["type"], "scheme": server_info["scheme"], "host": server_info["host"], "port": server_info["port"], "remotes": set(), "urls": 0, "auth_urls": 0, "plain_urls": 0, "tokens": [], "plain_remotes": [], }, ) entry["remotes"].add(remote_name) entry["urls"] += 1 for field_name in ["type", "scheme", "host", "port"]: if field_name not in entry or entry[field_name] in {"", None, "unknown"}: entry[field_name] = server_info[field_name] 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 token_id, token_value = credentials entry["tokens"].append( { "remote": remote_name, "org": server_info["org"], "repo": server_info["repo"], "token_id": token_id, "token_value": token_value, } ) return repo_root, repo_servers def compare_server_entries(repo_entry: dict | None, store_entry: dict | None) -> tuple[str, int, int]: if repo_entry is None and store_entry is None: return "missing", 0, 0 if repo_entry is not None and store_entry is None: return "repo_only", count_server_tokens(repo_entry), 0 if repo_entry is None and store_entry is not None: return "store_only", 0, count_server_tokens(store_entry) repo_pairs = token_pairs_from_entry(repo_entry) store_pairs = token_pairs_from_entry(store_entry) repo_only_pairs = repo_pairs - store_pairs store_only_pairs = store_pairs - repo_pairs if not repo_only_pairs and not store_only_pairs: return "in_sync", 0, 0 if repo_only_pairs and not store_only_pairs: return "repo_ahead", len(repo_only_pairs), 0 if store_only_pairs and not repo_only_pairs: return "store_ahead", 0, len(store_only_pairs) return "diverged", len(repo_only_pairs), len(store_only_pairs) def repo_result_for_entry(repo_entry: dict | None) -> tuple[str, str]: if repo_entry is None: return "missing", "missing" if repo_entry.get("auth_urls", 0): return "auth", "present" if repo_entry.get("plain_urls", 0): return "plain", "no_credentials" return "empty", "no_credentials" def print_token_context( repo_root: Path | None, config: WorkspaceConfig, repo_servers: dict[str, dict], store_servers: dict[str, dict], endpoint_count: int, ) -> None: print("context") print("item\tvalue") print(f"repo_root\t{repo_root or ''}") 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"tokens\t{endpoint_count}") def print_status_counts(status_counts: dict[str, int]) -> None: print("status") print("item\tvalue") for status_name in ["in_sync", "repo_only", "store_only", "repo_ahead", "store_ahead", "diverged"]: print(f"{status_name}\t{status_counts[status_name]}") TOKEN_REF_WIDTH = 9 TOKEN_COLUMNS = [ ("item", "item", 4), ("server", "server", 6), ("proto", "proto", 5), ("host", "host", 18), ("owner", "org", 9), ("repo_name", "repo", 11), ("user", "user", 4), ("remote", "remote", 6), ("token_ref", "token_ref", TOKEN_REF_WIDTH), ("token", "token", 12), ("valid", "valid", 19), ("scope_mask", "scope", 9), ("org_mask", "org", 6), ("repo_mask", "repo", 6), ] TOKEN_SEPARATOR_OVERRIDES = { "scope_mask": "aAimnopru", "org_mask": "oawrc-", "repo_mask": "oawr--", } def fit_cell(value: str, width: int) -> str: if len(value) <= width: return value.ljust(width) if width <= 3: return value[:width] return f"{value[:width - 3]}..." 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(fit_cell(TOKEN_SEPARATOR_OVERRIDES.get(name, "-" * width), width) for name, _, width in columns).rstrip()) for row in rows: print(" ".join(fit_cell(row.get(name, ""), width) for name, _, width in columns).rstrip()) 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: if server_entry is None: return "" return ",".join(sorted(str(user_name) for user_name in server_entry.get("users", {}))) def endpoint_column_values(endpoint: str, repo_entry: dict | None, store_entry: dict | None) -> dict[str, str]: values = { "server": "", "scheme": "", "host": "", "port": "", } for server_entry in [repo_entry, store_entry]: if server_entry is None: continue values["server"] = values["server"] or str(server_entry.get("type", "")) values["scheme"] = values["scheme"] or str(server_entry.get("scheme", "")) values["host"] = values["host"] or str(server_entry.get("host", "")) port_value = server_entry.get("port", "") values["port"] = values["port"] or ("" if port_value is None else str(port_value)) split_endpoint = urlsplit(endpoint) values["scheme"] = values["scheme"] or split_endpoint.scheme values["host"] = values["host"] or (split_endpoint.hostname or "") if not values["port"] and split_endpoint.port is not None: values["port"] = str(split_endpoint.port) if values["host"] and values["port"]: values["host"] = f"{values['host']}:{values['port']}" 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()): for row in server_entry.get("tokens", []): token_value = row.get("token_value", "") token_id = row.get("token_id", "") if token_id: rows.append( { "endpoint": endpoint, "user": str(row.get("user", "")), "token_id": str(token_id), "remote": str(row.get("remote", "")), "owner": str(row.get("owner", "")), "repo_name": str(row.get("repo_name", "")), "token_value": str(token_value), "expires_at": str(row.get("expires_at", "")), "valid": str(row.get("valid", "")), "scope_mask": str(row.get("scope_mask", "")), "org_mask": str(row.get("org_mask", "")), "repo_mask": str(row.get("repo_mask", "")), } ) 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", ""), "token_id": row.get("token_id", ""), "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", ""), "token_id": row.get("token_id", ""), "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} categories = [ "activitypub", "admin", "issue", "misc", "notification", "organization", "package", "repository", "user", ] values = [] for category in categories: if f"write:{category}" in scope_set: values.append("w") elif f"read:{category}" in scope_set: values.append("r") else: values.append("-") return "".join(values) 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, token_id: str, token_value: str, expires_at: str, owner: str, repo_name: str, ) -> dict: api_base = endpoint.rstrip("/") + "/api/v1" result = { "ok": False, "valid": "!", "scope_mask": "!!!!!!!!!", "org_mask": "!!!!!", "repo_mask": "!!!!", "user": "", } try: user_info = api_get_json(f"{api_base}/user", token_value) user_name = str(user_info.get("login", "")) if isinstance(user_info, dict) else "" result["user"] = user_name result["valid"] = valid_label(True, expires_at) if user_name: 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]) except (OSError, ValueError, urlerror.URLError): result["scope_mask"] = "?????????" if user_name and 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 user_name and 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 urlerror.HTTPError as error: result["valid"] = "invalid" if error.code in {401, 403} else "!" return result except (OSError, ValueError, urlerror.URLError): return result def print_token_item_rows( endpoint_names: list[str], repo_servers: dict[str, dict], store_servers: dict[str, dict], scan_rows: list[dict[str, str]] | None = None, ) -> 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} 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 store_rows = token_store_rows(store_servers) store_by_remote: dict[tuple[str, str], dict[str, str]] = {} for row in store_rows: if row["remote"]: store_by_remote.setdefault((row["endpoint"], row["remote"]), row) 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) used_store_keys: set[tuple[str, str, str]] = set() token_rows: list[dict[str, str]] = [] def append_token_row( endpoint: str, user_name: str, remote_name: str, token_id: str, token_value: str, marker: str, owner: str, repo_name: str, store_row: dict | None = None, ) -> None: endpoint_values = endpoint_values_by_name.get(endpoint, endpoint_column_values(endpoint, None, None)) valid_value = store_row.get("valid", "?") if store_row else "?" scope_value = store_row.get("scope_mask", "?????????") if store_row else "?????????" org_value = store_row.get("org_mask", "?????") if store_row else "?????" repo_value = store_row.get("repo_mask", "????") if store_row else "????" token_rows.append( { "item": str(len(token_rows) + 1), "server": endpoint_values["server"], "proto": endpoint_values["scheme"], "host": endpoint_values["host"], "owner": owner, "repo_name": repo_name, "user": user_name, "remote": remote_name, "token_ref": format_token_ref(token_id, marker), "token": short_secret(token_value), "valid": valid_value or "?", "scope_mask": scope_value or "?????????", "org_mask": org_value or "?????", "repo_mask": repo_value or "????", } ) for row in repo_token_rows(endpoint_names, repo_servers, scan_rows): endpoint = row["endpoint"] token_value = row["token_value"] store_row = store_by_remote.get((endpoint, row["remote"])) token_id = row["token_id"] expires_at = store_row.get("expires_at", "") if store_row else "" marker = "R" synced_store_row = None if store_row: repo_fields_match = ( store_row["token_id"] == row["token_id"] and store_row["token_value"] == token_value and store_row["owner"] == row["owner"] and store_row["repo_name"] == row["repo_name"] ) if repo_fields_match: used_store_keys.add((endpoint, store_row["token_id"], store_row["remote"])) synced_store_row = store_row marker = "!" if store_row.get("valid") in {"!", "invalid"} else "*" append_token_row( endpoint, synced_store_row.get("user", "") if synced_store_row else "", row["remote"], token_id, token_value, marker, row["owner"], row["repo_name"], synced_store_row, ) for row in store_rows: store_key = (row["endpoint"], row["token_id"], row["remote"]) if store_key in used_store_keys: continue fallback = project_fallback.get(row["endpoint"], {}) append_token_row( row["endpoint"], row["user"], row["remote"], row["token_id"], row["token_value"], "S", row["owner"] or fallback.get("owner", ""), row["repo_name"] or fallback.get("repo_name", ""), ) print_fixed_table("tokens", TOKEN_COLUMNS, token_rows) return status_counts def resolve_token_from_store( token_data: dict, endpoint: str | None, token_id: str | None, user_name: str | None, ) -> dict: candidates = [] for token_record in token_data.get("tokens", []): token_endpoint = token_server_endpoint(token_record) if endpoint and token_endpoint != endpoint: continue if token_id and token_record.get("token_id") != token_id: continue if user_name and token_record.get("user") != user_name: continue candidates.append(token_record) if not candidates: raise SystemExit("No matching token in tokens.json.") if len(candidates) > 1: available = ", ".join( sorted( f"{token_server_endpoint(token_record)}:{token_record.get('token_id', '')}" for token_record in candidates ) ) raise SystemExit(f"Multiple matching tokens. Pass --server/--token-name. Available: {available}") return candidates[0] def sanitize_remote_url(remote_url: str) -> tuple[str, bool]: split_url = urlsplit(remote_url) if split_url.scheme not in {"http", "https"} or split_url.hostname is None: raise SystemExit(f"Only http/https remote URLs are supported: {remote_url}") _, port_value = endpoint_from_split(split_url) default_port = default_port_for_scheme(split_url.scheme) netloc = split_url.hostname if port_value is not None and port_value != default_port: netloc = f"{netloc}:{port_value}" had_credentials = split_url.username is not None or split_url.password is not None sanitized_url = urlunsplit((split_url.scheme, netloc, split_url.path, split_url.query, split_url.fragment)) return sanitized_url, had_credentials def credentialed_remote_url(remote_url: str, user_name: str, token_value: str) -> str: sanitized_url, _ = sanitize_remote_url(remote_url) split_url = urlsplit(sanitized_url) _, port_value = endpoint_from_split(split_url) default_port = default_port_for_scheme(split_url.scheme) host_part = split_url.hostname or "" if port_value is not None and port_value != default_port: host_part = f"{host_part}:{port_value}" netloc = f"{quote(user_name, safe='')}:{quote(token_value, safe='')}@{host_part}" return urlunsplit((split_url.scheme, netloc, split_url.path, split_url.query, split_url.fragment)) def resolve_repo_argument(repo_arg: str | None) -> Path: return Path(repo_arg).expanduser().resolve() if repo_arg else Path.cwd().resolve() def run_tokens_scan(config: WorkspaceConfig, args: argparse.Namespace) -> None: repo_path = resolve_repo_argument(args.repo) token_data = load_token_store(config, write_normalized=False) store_servers = store_servers_from_tokens(token_data) report = scan_repo_remotes(config, repo_path) repo_root, repo_servers = collect_repo_server_entries(config, repo_path) endpoint_names = sorted(set(repo_servers) | set(store_servers)) print_token_item_rows(endpoint_names, repo_servers, store_servers, report["remote_rows"]) def run_tokens_read(config: WorkspaceConfig, args: argparse.Namespace) -> None: token_data = load_token_store(config, write_normalized=False) servers = token_store_servers(token_data) if args.server: server_entry = dict(store_servers_from_tokens(token_data)).get(args.server) if server_entry is None: raise SystemExit(f"Missing server endpoint in token store: {args.server}") servers = [(args.server, server_entry)] if not servers: print(f"token_path\t{config.token_path}") print("servers\t0") return print(f"token_path\t{config.token_path}") for index, (endpoint, server_entry) in enumerate(servers): if index: print() print(f"endpoint\t{endpoint}") print(f"type\t{server_entry.get('type', 'unknown')}") print(f"scheme\t{server_entry.get('scheme', '')}") print(f"host\t{server_entry.get('host', '')}") print(f"port\t{server_entry.get('port', '')}") print(f"tokens\t{count_server_tokens(server_entry)}") for row in sorted(server_entry.get("tokens", []), key=lambda item: (item.get("token_id", ""), item.get("remote", ""))): token_value = row.get("token_value", "") secret_value = token_value if args.show_secrets else mask_secret(token_value) print(f"token\t{row.get('token_id', '')}\t{secret_value}") for field_name in ["user", "valid", "scope_mask", "expires_at"]: field_value = row.get(field_name, "") if field_value: print(f"{field_name}\t{row.get('token_id', '')}\t{field_value}") if row.get("remote"): print( "remote\t" f"{row.get('token_id', '')}\t{row.get('remote', '')}\t" f"{row.get('owner', '')}\t{row.get('repo_name', '')}" ) for field_name in ["org_mask", "repo_mask"]: field_value = row.get(field_name, "") if field_value: print(f"{field_name}\t{row.get('token_id', '')}\t{field_value}") def run_tokens_stats(config: WorkspaceConfig, args: argparse.Namespace) -> None: repo_root, repo_servers = collect_repo_server_entries(config, resolve_repo_argument(args.repo)) token_data = load_token_store(config, write_normalized=False) store_servers = store_servers_from_tokens(token_data) endpoint_names = sorted(set(repo_servers) | set(store_servers)) if args.server: if args.server not in set(endpoint_names): raise SystemExit(f"Missing server endpoint in repo/store: {args.server}") endpoint_names = [args.server] print_token_context(repo_root, config, repo_servers, store_servers, len(endpoint_names)) print() status_counts = print_token_item_rows(endpoint_names, repo_servers, store_servers) print() print_status_counts(status_counts) def run_tokens_add(config: WorkspaceConfig, args: argparse.Namespace) -> None: server_url = (args.server or config.git.base_url).rstrip("/") server_info = server_info_from_url(config, server_url) if server_info is None: raise SystemExit(f"Only http/https server URLs are supported: {server_url}") token_data = load_token_store(config) existing_token = find_token_record(token_data, server_info["endpoint"], args.token_id) if existing_token is not None: raise SystemExit(f"Token already exists in tokens.json: {server_info['endpoint']}:{args.token_id}") token_record = { "token_id": args.token_id, "value": args.value or "", "server": server_record_from_info(server_info), "remotes": [], } if args.user: token_record["user"] = args.user if args.remote: token_record["remotes"].append( { "name": args.remote, "org": args.org or "", "repo": args.repo_name or "", } ) print(f"token_path\t{config.token_path}") print(f"server\t{server_info['endpoint']}") print(f"token_id\t{args.token_id}") print(f"value\t{'set' if token_record['value'] else 'empty'}") print(f"remote\t{args.remote or ''}") print(f"status\t{'dry-run' if args.dry_run else 'added'}") if args.dry_run: return token_data.setdefault("tokens", []).append(token_record) write_token_store(config, token_data) def run_tokens_write(config: WorkspaceConfig, args: argparse.Namespace) -> None: repo_path = resolve_repo_argument(args.repo) repo_root = git_repo_root(repo_path) if repo_root is None: raise SystemExit(f"Missing git repo at path: {repo_path}") if not args.remote: raise SystemExit("Pass --remote for tokens write.") token_data = load_token_store(config) existing_remote_url = git_capture(repo_root, ["remote", "get-url", args.remote]) target_url = args.url or existing_remote_url if not target_url: raise SystemExit("Missing target remote URL. Provide --url or create the remote first.") target_server_info = server_info_from_url(config, target_url) if target_server_info is None: raise SystemExit(f"Only http/https remote URLs are supported: {target_url}") token_record = resolve_token_from_store( token_data, args.server or target_server_info["endpoint"], args.token_name, args.user, ) server_endpoint = token_server_endpoint(token_record) token_id = str(token_record.get("token_id", "")) token_value = str(token_record.get("value", "")) if not token_value: raise SystemExit(f"Token value is empty in tokens.json: {server_endpoint}:{token_id}") final_url = credentialed_remote_url(target_url, token_id, token_value) status = "added" if existing_remote_url is None else "updated" existing_credentials = remote_credentials(existing_remote_url) if existing_remote_url else None if existing_credentials == (token_id, token_value): status = "unchanged" elif existing_credentials is not None and not args.replace: raise SystemExit( f"Remote '{args.remote}' already has different credentials. " "Pass --replace to overwrite them." ) print(f"repo_root\t{repo_root}") print(f"remote\t{args.remote}") print(f"server\t{server_endpoint}") print(f"user\t{token_record.get('user', '')}") print(f"token_name\t{token_id}") print(f"status\t{status}") print(f"url\t{sanitize_remote_url(final_url)[0]}") if args.dry_run or status == "unchanged": return set_git_remote(repo_root, args.remote, final_url) def run_tokens_update(config: WorkspaceConfig, args: argparse.Namespace) -> None: if args.from_source == "remotes": if args.dry_run: run_tokens_scan(config, args) return repo_path = resolve_repo_argument(args.repo) report = sync_tokens_from_repo(config, repo_path) repo_root, repo_servers = collect_repo_server_entries(config, repo_path) token_data = load_token_store(config, write_normalized=False) store_servers = store_servers_from_tokens(token_data) endpoint_names = sorted(set(repo_servers) | set(store_servers)) print_token_item_rows(endpoint_names, repo_servers, store_servers, report["remote_rows"]) return if args.from_source == "store": run_tokens_write(config, args) return raise SystemExit(f"Unsupported tokens update source: {args.from_source}") def git_current_branch(config: WorkspaceConfig, repo_path: Path) -> str: branch_name = git_capture(repo_path, ["branch", "--show-current"]) if branch_name: return branch_name head_ref = git_capture(repo_path, ["symbolic-ref", "--short", f"refs/remotes/{config.git.origin_remote}/HEAD"]) if head_ref and "/" in head_ref: return head_ref.split("/", 1)[1] return config.git.fallback_branch def fallback_source_repo_name(series_name: str, card_name: str) -> str: return "-".join([slug_value(series_name, "series"), slug_value(card_name, "card")]) def fallback_source_url(config: WorkspaceConfig, series_name: str, card_name: str) -> str: repo_name = fallback_source_repo_name(series_name, card_name) return f"{config.git.base_url}/{config.git.source_org}/{repo_name}.git" def discover_source_url(config: WorkspaceConfig, repo_path: Path, series_name: str, card_name: str) -> str: source_url = git_capture(repo_path, ["remote", "get-url", config.git.origin_remote]) if source_url: return source_url source_url = git_capture(repo_path, ["remote", "get-url", config.git.source_remote]) if source_url: return source_url return fallback_source_url(config, series_name, card_name) def repo_name_from_url(remote_url: str) -> str: repo_name = remote_url.rstrip("/").rsplit("/", 1)[-1] if repo_name.endswith(".git"): repo_name = repo_name[:-4] return slug_value(repo_name, "source repo") def build_answer_repo_name(source_repo_name: str, class_name: str, date_value: str) -> str: return "-".join([source_repo_name, slug_value(class_name, "class"), date_value]) def build_answer_url(config: WorkspaceConfig, answer_repo_name: str) -> str: return f"{config.git.base_url}/{config.git.answer_org}/{answer_repo_name}.git" def set_git_remote(repo_path: Path, remote_name: str, remote_url: str) -> str: existing_url = git_capture(repo_path, ["remote", "get-url", remote_name]) if existing_url == remote_url: return "unchanged" if existing_url: subprocess.run( ["git", "-C", str(repo_path), "remote", "set-url", remote_name, remote_url], check=True, ) return "updated" subprocess.run( ["git", "-C", str(repo_path), "remote", "add", remote_name, remote_url], check=True, ) return "added" def print_submission_plan(config: WorkspaceConfig, args: argparse.Namespace) -> None: series_name, card_name = resolve_selector(config, args.series, args.card) selector = f"{series_name}/{card_name}" card_path = resolve_card_path(config, series_name, card_name) sync_tokens_from_repo(config, card_path) class_name = slug_value(args.class_name, "class") branch_name = slug_value(args.branch or args.nick, "branch") source_url = args.source_url or discover_source_url(config, card_path, series_name, card_name) source_repo_name = repo_name_from_url(source_url) answer_repo_name = build_answer_repo_name(source_repo_name, class_name, args.date) answer_url = build_answer_url(config, answer_repo_name) source_branch = git_current_branch(config, card_path) print(f"selector\t{selector}") print(f"card_path\t{card_path}") print(f"source_repo\t{config.git.source_org}/{source_repo_name}") print(f"source_remote\t{config.git.source_remote}") print(f"source_url\t{source_url}") print(f"source_branch\t{source_branch}") print(f"answer_remote\t{config.git.answer_remote}") print(f"answer_repo\t{config.git.answer_org}/{answer_repo_name}") print(f"answer_url\t{answer_url}") print(f"student_branch\t{branch_name}") if args.apply: source_status = set_git_remote(card_path, config.git.source_remote, source_url) answer_status = set_git_remote(card_path, config.git.answer_remote, answer_url) print(f"apply_{config.git.source_remote}\t{source_status}") print(f"apply_{config.git.answer_remote}\t{answer_status}") print("commands") print(f" git -C {shlex.quote(str(card_path))} remote add {config.git.source_remote} {shlex.quote(source_url)}") print(f" git -C {shlex.quote(str(card_path))} remote add {config.git.answer_remote} {shlex.quote(answer_url)}") print(f" git -C {shlex.quote(str(card_path))} fetch {config.git.source_remote} {shlex.quote(source_branch)}") print(f" git -C {shlex.quote(str(card_path))} switch -c {shlex.quote(branch_name)}") print(f" git -C {shlex.quote(str(card_path))} push -u {config.git.answer_remote} {shlex.quote(branch_name)}") def tmux_target_exists(session_name: str) -> bool: result = subprocess.run( ["tmux", "has-session", "-t", session_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) return result.returncode == 0 def build_container_command( config: WorkspaceConfig, selector: str, card_path: Path, instance: str, ) -> str: tool_root = config.tools_root rv_script = tool_root / "rv" if not rv_script.is_file(): candidates = "\n".join(f" - {path}" for path in config.tools_root_candidates) raise SystemExit(f"Missing rv launcher. Checked:\n{candidates}") env_parts = [ f"WORKSPACE_ROOT={shlex.quote(str(config.workspace_root))}", f"RV_REPO_PATH={shlex.quote(str(card_path))}", f"RV_CARD={shlex.quote(selector)}", f"RV_INSTANCE={shlex.quote(instance)}", ] return " && ".join( [ f"cd {shlex.quote(str(tool_root))}", " ".join(env_parts + ["bash", "./rv", "shell"]), ] ) def run_tmux_container(config: WorkspaceConfig, args: argparse.Namespace) -> None: series_name, card_name = resolve_selector(config, args.series, args.card) selector = f"{series_name}/{card_name}" card_path = resolve_card_path(config, series_name, card_name) session_name = args.session or config.defaults.get("tmux_session", "rv-workspace") window_name = args.window or config.defaults.get("tmux_window", "rv") instance = args.instance or config.defaults.get("instance", "shell") if tmux_target_exists(session_name): raise SystemExit( f"tmux session already exists: {session_name}\n" "Use another --session name or remove the existing session first." ) shell_command = build_container_command(config, selector, card_path, instance) tmux_command = [ "tmux", "new-session", "-d", "-s", session_name, "-n", window_name, f"bash -lc {shlex.quote(shell_command)}", ] if args.dry_run: print("Selector:", selector) print("Card path:", card_path) print("Tools root:", config.tools_root) print("Pane 0 command:") print(shell_command) print("tmux command:") print(shlex.join(tmux_command)) return subprocess.run(tmux_command, check=True) if args.attach: subprocess.run(["tmux", "attach", "-t", session_name], check=True) return print(f"Created tmux session: {session_name}") print(f"Window: {window_name}") print("Pane 0: container shell") print(f"Attach with: tmux attach -t {session_name}") def print_config(config: WorkspaceConfig) -> None: print(f"config_path\t{config.config_path}") print(f"original_root\t{config.original_root}") print(f"original_series_root\t{config.original_series_root}") print(f"workspace_root\t{config.workspace_root}") print(f"series_root\t{config.series_root}") print(f"socket_root\t{config.socket_root}") print(f"token_path\t{config.token_path}") print(f"tools_root\t{config.tools_root}") print(f"git_base_url\t{config.git.base_url}") print(f"git_source_org\t{config.git.source_org}") print(f"git_answer_org\t{config.git.answer_org}") print(f"git_source_remote\t{config.git.source_remote}") print(f"git_answer_remote\t{config.git.answer_remote}") print(f"git_origin_remote\t{config.git.origin_remote}") print(f"git_fallback_branch\t{config.git.fallback_branch}") print("tools_root_candidates") for path in config.tools_root_candidates: print(f" {path}") def add_repo_option(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--repo", help="Path inside a target git repo. Default: current working directory.", ) def add_store_selection_options(parser: argparse.ArgumentParser) -> None: parser.add_argument("--server", help="Server endpoint from tokens.json.") parser.add_argument("--user", help="API user recorded on the token.") parser.add_argument("--token-name", help="Token id inside tokens.json, for example 't1'.") def print_table(headers: list[str], rows: list[list[str]]) -> None: widths = [len(header) for header in headers] for row in rows: for index, value in enumerate(row): widths[index] = max(widths[index], len(value)) print(" ".join(header.ljust(widths[index]) for index, header in enumerate(headers))) print(" ".join("-" * width for width in widths)) for row in rows: print(" ".join(value.ljust(widths[index]) for index, value in enumerate(row))) def print_config_summary(config_path: Path) -> None: try: config = load_config(config_path) except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as error: print("Config:") print_table( ["field", "value"], [ ["config", str(config_path)], ["status", f"unavailable: {error}"], ], ) return print("Config:") print_table( ["field", "value"], [ ["config", str(config.config_path)], ["workspace", str(config.workspace_root)], ["series", str(config.series_root)], ["tokens", str(config.token_path)], ], ) def print_main_overview(config_path: Path) -> None: print("RVCTL - launcher workspace RV") print() print("Usage:") print(" ./rvctl [options]") print(" ./rvctl tokens [options]") print(" ./rvctl --help") print() print("Commands:") print_table( ["command", "arguments", "purpose"], [ ["show-config", "", "show resolved paths and Git defaults"], ["list-series", "", "list series from workspace"], ["list-cards", "[series]", "list cards in a selected series"], ["tmux-container", "[series] [card]", "start tmux with container in pane 0"], ["submission", "[series] [card] --class K --nick N", "prepare answer repo and student branch"], ["tokens", "scan|add|read|stats|write|update", "manage tokens.json and Git remote credentials"], ], ) print() print("Typical flow:") print_table( ["step", "command", "result"], [ ["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 ", "start the working container"], ["5", "./rvctl submission inf --class 4i --nick u1", "prepare answer remote and branch"], ], ) print() print_config_summary(config_path) print() print("Docs:") print(" doc/rvctl.md") print(" doc/tokens.md") print(" doc/tokens.schema.json") print(" doc/workspace.md") def print_tokens_overview() -> None: print("RVCTL tokens - token store and Git remote credentials") print() print("Usage:") print(" ./rvctl tokens [options]") print(" ./rvctl tokens --help") print() print("Commands:") print_table( ["command", "common options", "direction", "purpose"], [ ["scan", "[--repo PATH]", "read-only", "print token table with remote/store marker and auth masks"], ["add", "TOKEN_ID", "tokens.json", "add an empty token skeleton for manual editing"], ["read", "[--server ENDPOINT]", "tokens.json", "show servers, token ids and remotes"], ["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"], ["update", "--from remotes|store", "selected", "copy token data in an explicit direction"], ], ) print() print("Examples:") print_table( ["case", "command"], [ ["scan current repo", "./rvctl tokens scan"], ["scan selected card", "./rvctl tokens scan --repo ~/dev/workspace/rv/series/inf/03"], ["compare state", "./rvctl tokens stats --repo ~/dev/workspace/rv/series/inf/03"], ["write auth to r1", "./rvctl tokens write --repo PATH --remote r1 --server URL --token-name t1"], ], ) def extract_config_path(argv: list[str]) -> tuple[Path, list[str]]: config_path = Path(__file__).with_name("workspace.json") remaining: list[str] = [] index = 0 while index < len(argv): arg = argv[index] if arg == "--config": if index + 1 >= len(argv): remaining.append(arg) index += 1 continue config_path = Path(argv[index + 1]) index += 2 continue if arg.startswith("--config="): config_path = Path(arg.split("=", 1)[1]) index += 1 continue remaining.append(arg) index += 1 return config_path, remaining def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="rvctl", description="rvctl for rv cards and tools.", ) parser.add_argument( "--config", default=Path(__file__).with_name("workspace.json"), type=Path, help="Path to workspace.json", ) subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("show-config", help="Print resolved workspace paths.") subparsers.add_parser("list-series", help="List series directories from series_root.") cards_parser = subparsers.add_parser("list-cards", help="List cards in a series.") cards_parser.add_argument("series", nargs="?", help="Series id, for example 'inf'.") tmux_parser = subparsers.add_parser( "tmux-container", help="Create a tmux session with pane 0 running the container shell for a selected card.", ) tmux_parser.add_argument("series", nargs="?", help="Series id or full selector like 'inf/03'.") tmux_parser.add_argument("card", nargs="?", help="Card id, for example '03'.") tmux_parser.add_argument("--session", help="tmux session name.") tmux_parser.add_argument("--window", help="tmux window name.") tmux_parser.add_argument("--instance", help="RV_INSTANCE passed to the tool launcher.") tmux_parser.add_argument("--attach", action="store_true", help="Attach to the session after creation.") tmux_parser.add_argument("--dry-run", action="store_true", help="Print commands instead of running tmux.") submission_parser = subparsers.add_parser( "submission", help="Print or apply source/answer git remotes for a selected card.", ) submission_parser.add_argument("series", nargs="?", help="Series id or full selector like 'inf/03'.") submission_parser.add_argument("card", nargs="?", help="Card id, for example '03'.") submission_parser.add_argument("--class", dest="class_name", required=True, help="Class id, for example '4i'.") submission_parser.add_argument("--nick", required=True, help="Student nick used as the branch name.") submission_parser.add_argument( "--branch", help="Override the submission branch name. Defaults to the value of --nick.", ) submission_parser.add_argument( "--date", default=dt.date.today().isoformat(), help="Lesson date used in the answer repo name. Default: today in YYYY-MM-DD.", ) submission_parser.add_argument( "--source-url", help="Override the source repo URL. By default launcher reads the current origin from the card repo.", ) submission_parser.add_argument( "--apply", action="store_true", help="Write remotes into the selected card repo instead of only printing the plan.", ) tokens_parser = subparsers.add_parser("tokens", help="Manage local token store and repo remotes.") tokens_subparsers = tokens_parser.add_subparsers(dest="tokens_command", required=True) tokens_scan_parser = tokens_subparsers.add_parser( "scan", help="Read repo remotes and tokens.json, then print paired token rows.", ) add_repo_option(tokens_scan_parser) tokens_read_parser = tokens_subparsers.add_parser( "read", help="Read tokens.json and print stored servers, token ids and remotes.", ) tokens_read_parser.add_argument("--server", help="Filter output to one server endpoint.") tokens_read_parser.add_argument("--show-secrets", action="store_true", help="Print full token values.") tokens_add_parser = tokens_subparsers.add_parser( "add", help="Add an empty token skeleton to tokens.json.", ) tokens_add_parser.add_argument("token_id", help="Token id, for example 't1' or 'r6'.") tokens_add_parser.add_argument("--server", help="Server endpoint. Default: git.base_url from workspace.json.") tokens_add_parser.add_argument("--value", default="", help="Optional token value. Default: empty for manual editing.") tokens_add_parser.add_argument("--user", help="Optional API user label.") tokens_add_parser.add_argument("--remote", help="Optional remote name to attach to this token.") tokens_add_parser.add_argument("--org", help="Optional remote organization.") tokens_add_parser.add_argument("--repo", dest="repo_name", help="Optional remote repository.") tokens_add_parser.add_argument("--dry-run", action="store_true", help="Print the planned token without writing it.") tokens_stats_parser = tokens_subparsers.add_parser( "stats", help="Print per-server token statistics from tokens.json.", ) add_repo_option(tokens_stats_parser) tokens_stats_parser.add_argument("--server", help="Filter output to one server endpoint.") tokens_write_parser = tokens_subparsers.add_parser( "write", help="Write credentials from tokens.json into a repo remote URL.", ) add_repo_option(tokens_write_parser) tokens_write_parser.add_argument("--remote", help="Remote name to update or create.") tokens_write_parser.add_argument("--url", help="Target remote URL if the remote does not exist yet.") add_store_selection_options(tokens_write_parser) tokens_write_parser.add_argument("--replace", action="store_true", help="Replace different credentials in the remote URL.") tokens_write_parser.add_argument("--dry-run", action="store_true", help="Print the planned update without writing it.") tokens_update_parser = tokens_subparsers.add_parser( "update", help="Synchronize token data in a selected direction.", ) tokens_update_parser.add_argument( "--from", dest="from_source", required=True, choices=["remotes", "store"], help="Synchronization direction source: 'remotes' or 'store'.", ) add_repo_option(tokens_update_parser) tokens_update_parser.add_argument("--remote", help="Remote name used when --from store.") tokens_update_parser.add_argument("--url", help="Target remote URL used when --from store.") add_store_selection_options(tokens_update_parser) tokens_update_parser.add_argument("--replace", action="store_true", help="Replace different credentials when writing to remotes.") tokens_update_parser.add_argument("--dry-run", action="store_true", help="Print the planned update without writing it.") return parser def main() -> int: config_path, command_args = extract_config_path(sys.argv[1:]) if not command_args or command_args == ["help"]: print_main_overview(config_path) return 0 if command_args in (["tokens"], ["help", "tokens"], ["tokens", "help"]): print_tokens_overview() return 0 parser = build_parser() args = parser.parse_args() config = load_config(args.config) if args.command == "show-config": print_config(config) return 0 if args.command == "list-series": print_series(config) return 0 if args.command == "list-cards": print_cards(config, args.series) return 0 if args.command == "tmux-container": run_tmux_container(config, args) return 0 if args.command == "submission": print_submission_plan(config, args) return 0 if args.command == "tokens": if args.tokens_command == "scan": run_tokens_scan(config, args) return 0 if args.tokens_command == "add": run_tokens_add(config, args) return 0 if args.tokens_command == "read": run_tokens_read(config, args) return 0 if args.tokens_command == "stats": run_tokens_stats(config, args) return 0 if args.tokens_command == "write": run_tokens_write(config, args) return 0 if args.tokens_command == "update": run_tokens_update(config, args) return 0 parser.error(f"Unsupported tokens command: {args.tokens_command}") return 2 parser.error(f"Unsupported command: {args.command}") return 2 if __name__ == "__main__": sys.exit(main())