mpabi
2004 lines
74 KiB
Python
2004 lines
74 KiB
Python
#!/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 '<series> <card>' or '<series/card>', 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": 2, "servers": {}}
|
|
|
|
|
|
def next_token_name(tokens: dict[str, str]) -> str:
|
|
index = 1
|
|
while f"t{index}" in tokens:
|
|
index += 1
|
|
return f"t{index}"
|
|
|
|
|
|
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 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 ensure_server_entry(token_data: dict, server_info: dict) -> dict:
|
|
servers = token_data.setdefault("servers", {})
|
|
endpoint = server_info["endpoint"]
|
|
server_entry = servers.setdefault(
|
|
endpoint,
|
|
{
|
|
"type": server_info["type"],
|
|
"scheme": server_info["scheme"],
|
|
"host": server_info["host"],
|
|
"port": server_info["port"],
|
|
"users": {},
|
|
},
|
|
)
|
|
server_entry.setdefault("users", {})
|
|
for field_name in ["type", "scheme", "host", "port"]:
|
|
if field_name not in server_entry or server_entry[field_name] in {"", None, "unknown"}:
|
|
server_entry[field_name] = server_info[field_name]
|
|
return server_entry
|
|
|
|
|
|
def copy_legacy_users(token_data: dict, server_info: dict, raw_users: dict) -> dict:
|
|
server_entry = ensure_server_entry(token_data, server_info)
|
|
server_users = server_entry.setdefault("users", {})
|
|
|
|
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
|
|
|
|
target_user = server_users.setdefault(str(user_name), {"tokens": {}})
|
|
target_tokens = target_user.setdefault("tokens", {})
|
|
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)
|
|
if token_value_exists(target_tokens, token_value):
|
|
continue
|
|
|
|
normalized_name = str(token_name)
|
|
if normalized_name in target_tokens and token_entry_value(target_tokens[normalized_name]) != token_value:
|
|
normalized_name = next_token_name(target_tokens)
|
|
target_tokens[normalized_name] = token_entry
|
|
|
|
return token_data
|
|
|
|
|
|
def normalize_token_store(config: WorkspaceConfig, raw_data: dict) -> dict:
|
|
if "servers" in raw_data and isinstance(raw_data["servers"], dict):
|
|
token_data = empty_token_store()
|
|
token_data["version"] = raw_data.get("version", 2)
|
|
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"),
|
|
}
|
|
target_server = ensure_server_entry(token_data, base_server_info)
|
|
copy_legacy_users(token_data, base_server_info, server_entry.get("users", {}))
|
|
target_server["users"] = token_data["servers"][str(endpoint)]["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_entry_with_metadata(token_value: str, server_info: dict, remote_name: str) -> dict[str, str]:
|
|
token_entry = {"value": token_value}
|
|
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:
|
|
token_entry[target_name] = source_value
|
|
if remote_name:
|
|
token_entry["remote"] = remote_name
|
|
return token_entry
|
|
|
|
|
|
def update_token_entry_metadata(token_entry, token_value: str, server_info: dict, remote_name: str) -> tuple[dict, bool]:
|
|
if isinstance(token_entry, dict):
|
|
updated_entry = dict(token_entry)
|
|
else:
|
|
updated_entry = {"value": token_value}
|
|
|
|
changed = updated_entry.get("value") != token_value
|
|
updated_entry["value"] = token_value
|
|
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 updated_entry.get(target_name) != source_value:
|
|
updated_entry[target_name] = source_value
|
|
changed = True
|
|
if remote_name and updated_entry.get("remote") != remote_name:
|
|
updated_entry["remote"] = remote_name
|
|
changed = True
|
|
return updated_entry, changed
|
|
|
|
|
|
def register_token(token_data: dict, server_info: dict, remote_name: str, user_name: str, token_value: str) -> str:
|
|
server_entry = ensure_server_entry(token_data, server_info)
|
|
users = server_entry.setdefault("users", {})
|
|
user_entry = users.setdefault(user_name, {"tokens": {}})
|
|
tokens = user_entry.setdefault("tokens", {})
|
|
|
|
for token_name, token_entry in list(tokens.items()):
|
|
if token_entry_field(token_entry, "remote") != remote_name:
|
|
continue
|
|
updated_entry, changed = update_token_entry_metadata(token_entry, token_value, server_info, remote_name)
|
|
tokens[token_name] = updated_entry
|
|
return "updated" if changed else "existing"
|
|
|
|
for token_name, token_entry in list(tokens.items()):
|
|
if token_entry_value(token_entry) != token_value:
|
|
continue
|
|
updated_entry, changed = update_token_entry_metadata(token_entry, token_value, server_info, remote_name)
|
|
tokens[token_name] = updated_entry
|
|
return "updated" if changed else "existing"
|
|
|
|
tokens[next_token_name(tokens)] = token_entry_with_metadata(token_value, server_info, remote_name)
|
|
return "added"
|
|
|
|
|
|
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",
|
|
"user": "",
|
|
"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",
|
|
"user": "",
|
|
"token_value": "",
|
|
"result": "no_credentials",
|
|
}
|
|
)
|
|
continue
|
|
|
|
auth_urls += 1
|
|
user_name, 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",
|
|
"user": user_name,
|
|
"token_value": token_value,
|
|
"result": "found",
|
|
}
|
|
)
|
|
found_credentials.append((row_index, server_info, remote_name, user_name, 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, user_name, token_value in found_credentials:
|
|
result = register_token(token_data, server_info, remote_name, user_name, 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 sum(len(user_entry.get("tokens", {})) for user_entry in server_entry.get("users", {}).values())
|
|
|
|
|
|
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 token_store_servers(token_data: dict) -> list[tuple[str, dict]]:
|
|
servers = token_data.get("servers", {})
|
|
return sorted(servers.items())
|
|
|
|
|
|
def token_pairs_from_users(users_data: dict) -> set[tuple[str, str, str]]:
|
|
token_pairs: set[tuple[str, str, str]] = set()
|
|
for user_name, user_entry in users_data.items():
|
|
if not isinstance(user_entry, dict):
|
|
continue
|
|
for token_entry in user_entry.get("tokens", {}).values():
|
|
token_value = token_entry_value(token_entry)
|
|
if token_value:
|
|
token_pairs.add((token_entry_field(token_entry, "remote"), str(user_name), token_value))
|
|
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,
|
|
"users": {},
|
|
"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
|
|
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 not token_value_exists(tokens, token_value):
|
|
tokens[next_token_name(tokens)] = token_entry_with_metadata(token_value, server_info, remote_name)
|
|
|
|
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_users(repo_entry.get("users", {}))
|
|
store_pairs = token_pairs_from_users(store_entry.get("users", {}))
|
|
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 token_name_rows(users_data: dict) -> list[tuple[str, str]]:
|
|
rows: list[tuple[str, str]] = []
|
|
for user_name in sorted(users_data):
|
|
user_entry = users_data[user_name]
|
|
if not isinstance(user_entry, dict):
|
|
continue
|
|
tokens = user_entry.get("tokens", {})
|
|
if not tokens:
|
|
rows.append((str(user_name), ""))
|
|
continue
|
|
for token_name in sorted(tokens):
|
|
rows.append((str(user_name), str(token_name)))
|
|
return rows
|
|
|
|
|
|
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", 6),
|
|
("org_mask", "org", 6),
|
|
("repo_mask", "repo", 6),
|
|
]
|
|
|
|
TOKEN_SEPARATOR_OVERRIDES = {
|
|
"scope_mask": "awrop-",
|
|
"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()):
|
|
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_entry in sorted(user_entry.get("tokens", {}).items()):
|
|
token_value = token_entry_value(token_entry)
|
|
if token_value:
|
|
rows.append(
|
|
{
|
|
"endpoint": endpoint,
|
|
"user": str(user_name),
|
|
"token_name": str(token_name),
|
|
"remote": token_entry_field(token_entry, "remote"),
|
|
"owner": token_entry_field(token_entry, "org"),
|
|
"repo_name": token_entry_field(token_entry, "repo"),
|
|
"token_value": token_value,
|
|
"expires_at": token_entry_expires_at(token_entry),
|
|
}
|
|
)
|
|
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,
|
|
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": "!!!!",
|
|
}
|
|
|
|
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)
|
|
result["valid"] = valid_label(True, expires_at)
|
|
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 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)
|
|
authz_cache: dict[tuple[str, 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"],
|
|
"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_name, marker),
|
|
"token": short_secret(token_value),
|
|
"valid": authz.get("valid", "?") if authz else "?",
|
|
"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_remote.get((endpoint, row["remote"]))
|
|
token_name = store_row["token_name"] if store_row else ""
|
|
expires_at = store_row.get("expires_at", "") if store_row else ""
|
|
marker = "R"
|
|
authz = None
|
|
if store_row:
|
|
used_store_keys.add((endpoint, store_row["user"], store_row["token_name"], store_row["token_value"]))
|
|
if store_row["user"] == row["user"] and store_row["token_value"] == token_value:
|
|
authz_key = (endpoint, row["user"], token_value, expires_at, 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,
|
|
expires_at,
|
|
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["remote"],
|
|
row["token_name"],
|
|
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_server_from_store(token_data: dict, endpoint: str | None) -> tuple[str, dict]:
|
|
servers = token_data.get("servers", {})
|
|
if endpoint:
|
|
server_entry = servers.get(endpoint)
|
|
if server_entry is None:
|
|
raise SystemExit(f"Missing server endpoint in token store: {endpoint}")
|
|
return endpoint, server_entry
|
|
|
|
if not servers:
|
|
raise SystemExit("Token store is empty.")
|
|
if len(servers) > 1:
|
|
available = ", ".join(sorted(servers))
|
|
raise SystemExit(f"Multiple servers in token store. Pass --server. Available: {available}")
|
|
|
|
endpoint_value = next(iter(sorted(servers)))
|
|
return endpoint_value, servers[endpoint_value]
|
|
|
|
|
|
def resolve_user_from_server(server_entry: dict, user_name: str | None) -> tuple[str, dict]:
|
|
users = server_entry.get("users", {})
|
|
if user_name:
|
|
user_entry = users.get(user_name)
|
|
if user_entry is None:
|
|
raise SystemExit(f"Missing user in token store: {user_name}")
|
|
return user_name, user_entry
|
|
|
|
if not users:
|
|
raise SystemExit("No users found in the selected server entry.")
|
|
if len(users) > 1:
|
|
available = ", ".join(sorted(users))
|
|
raise SystemExit(f"Multiple users in selected server entry. Pass --user. Available: {available}")
|
|
|
|
user_value = next(iter(sorted(users)))
|
|
return user_value, users[user_value]
|
|
|
|
|
|
def resolve_token_from_user(user_entry: dict, token_name: str | None) -> tuple[str, str]:
|
|
tokens = user_entry.get("tokens", {})
|
|
if token_name:
|
|
token_entry = tokens.get(token_name)
|
|
token_value = token_entry_value(token_entry)
|
|
if not token_value:
|
|
raise SystemExit(f"Missing token name in token store: {token_name}")
|
|
return token_name, token_value
|
|
|
|
if not tokens:
|
|
raise SystemExit("No tokens found for the selected user.")
|
|
if len(tokens) > 1:
|
|
available = ", ".join(sorted(tokens))
|
|
raise SystemExit(f"Multiple tokens for the selected user. Pass --token-name. Available: {available}")
|
|
|
|
token_key = next(iter(sorted(tokens)))
|
|
return token_key, token_entry_value(tokens[token_key])
|
|
|
|
|
|
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 = token_data.get("servers", {})
|
|
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 = token_data.get("servers", {}).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"users\t{len(server_entry.get('users', {}))}")
|
|
print(f"tokens\t{count_server_tokens(server_entry)}")
|
|
for user_name in sorted(server_entry.get("users", {})):
|
|
user_entry = server_entry["users"][user_name]
|
|
print(f"user\t{user_name}")
|
|
for token_name in sorted(user_entry.get("tokens", {})):
|
|
token_entry = user_entry["tokens"][token_name]
|
|
token_value = token_entry_value(token_entry)
|
|
secret_value = token_value if args.show_secrets else mask_secret(token_value)
|
|
print(f"token\t{token_name}\t{secret_value}")
|
|
for field_name in ["remote", "org", "repo"]:
|
|
field_value = token_entry_field(token_entry, field_name)
|
|
if field_value:
|
|
print(f"{field_name}\t{token_name}\t{field_value}")
|
|
expires_at = token_entry_expires_at(token_entry)
|
|
if expires_at:
|
|
print(f"expires_at\t{token_name}\t{expires_at}")
|
|
|
|
|
|
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 = token_data.get("servers", {})
|
|
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_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}")
|
|
|
|
server_endpoint, server_entry = resolve_server_from_store(token_data, args.server or target_server_info["endpoint"])
|
|
user_name, user_entry = resolve_user_from_server(server_entry, args.user)
|
|
token_name, token_value = resolve_token_from_user(user_entry, args.token_name)
|
|
|
|
final_url = credentialed_remote_url(target_url, user_name, 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 == (user_name, 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{user_name}")
|
|
print(f"token_name\t{token_name}")
|
|
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 = token_data.get("servers", {})
|
|
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="User name inside the selected server entry.")
|
|
parser.add_argument("--token-name", help="Token name inside the selected user entry, 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 <command> [options]")
|
|
print(" ./rvctl tokens <command> [options]")
|
|
print(" ./rvctl <command> --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|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 <card>", "start the working container"],
|
|
["5", "./rvctl submission inf <card> --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/workspace.md")
|
|
|
|
|
|
def print_tokens_overview() -> None:
|
|
print("RVCTL tokens - token store and Git remote credentials")
|
|
print()
|
|
print("Usage:")
|
|
print(" ./rvctl tokens <command> [options]")
|
|
print(" ./rvctl tokens <command> --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"],
|
|
["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"],
|
|
["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 --user u1 --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, users and token names.",
|
|
)
|
|
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_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 == "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())
|