Something went wrong. Try again.
a pds written in elixir tempest.desertthunder.dev
Something went wrong. Try again.
28 kB · 731 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732"""Tempest account migration CLI."""
from __future__ import annotations
import argparseimport jsonimport mimetypesimport osimport sysimport timeimport urllib.parsefrom dataclasses import dataclassfrom enum import StrEnumfrom pathlib import Pathfrom typing import Any
import httpx
from tempest_py import ar
DEFAULT_DID = "did:plc:oga6ppys7zwxlheuqmcm7dac"DEFAULT_HANDLE = "tempestpds.bsky.social"DEFAULT_OLD_PDS = "https://jellybaby.us-east.host.bsky.network"DEFAULT_TEMPEST = "https://tempest.desertthunder.dev"DEFAULT_TEMPEST_SERVICE_DID = "did:web:tempest.desertthunder.dev"CREATE_ACCOUNT_LXM = "com.atproto.server.createAccount"ARGON_COMMANDS = {"ar", "arg2", "argon"}
class CliError(RuntimeError): pass
class Command(StrEnum): FULL = "full" LOGIN_SOURCE = "login-source" SERVICE_AUTH = "service-auth" SOURCE_SESSION_STATUS = "source-session-status" EXPORT_CAR = "export-car" LIST_SOURCE_BLOBS = "list-source-blobs" DOWNLOAD_SOURCE_BLOBS = "download-source-blobs" CREATE_ACCOUNT = "create-account" REFRESH_SESSION = "refresh-session" IMPORT_REPO = "import-repo" STATUS = "status" MISSING_BLOBS = "missing-blobs" UPLOAD_MISSING_BLOBS = "upload-missing-blobs" PLC_RECOMMENDED = "plc-recommended" PLC_REQUEST_TOKEN = "plc-request-token" PLC_SIGN = "plc-sign" PLC_SUBMIT = "plc-submit" ACTIVATE = "activate"
@dataclass(frozen=True)class Settings: artifact_dir: Path old_pds: str old_login_pds: str old_auth_pds: str tempest: str tempest_service_did: str did: str handle: str email: str | None old_password: str | None tempest_password: str | None old_session_path: Path service_auth_path: Path car_path: Path source_blobs_path: Path create_account_path: Path import_repo_path: Path status_path: Path missing_blobs_path: Path plc_recommended_path: Path plc_token_path: Path plc_signed_path: Path plc_submit_path: Path activate_path: Path
def env(name: str, default: str | None = None) -> str | None: value = os.environ.get(name) return value if value not in (None, "") else default
def require_env(settings: Settings, name: str, value: str | None) -> str: if value: return value raise CliError(f"{name} is required for this command")
def path_from_env(name: str, default: Path) -> Path: value = env(name) return Path(value) if value else default
def settings_from_env(args: argparse.Namespace) -> Settings: artifact_dir = Path(args.artifact_dir or env("ARTIFACT_DIR", ".sandbox"))
return Settings( artifact_dir=artifact_dir, old_pds=env("OLD_PDS", DEFAULT_OLD_PDS).rstrip("/"), old_auth_pds=env("OLD_AUTH_PDS", env("OLD_PDS", DEFAULT_OLD_PDS)).rstrip("/"), old_login_pds=env("OLD_LOGIN_PDS", env("OLD_AUTH_PDS", env("OLD_PDS", DEFAULT_OLD_PDS))).rstrip("/"), tempest=env("TEMPEST", DEFAULT_TEMPEST).rstrip("/"), tempest_service_did=env("TEMPEST_SERVICE_DID", DEFAULT_TEMPEST_SERVICE_DID), did=env("DID", DEFAULT_DID), handle=env("HANDLE", DEFAULT_HANDLE), email=env("EMAIL"), old_password=env("OLD_PASSWORD"), tempest_password=env("TEMPEST_PASSWORD"), old_session_path=path_from_env("OLD_SESSION_JSON", artifact_dir / "old_session.json"), service_auth_path=path_from_env("SERVICE_AUTH_JSON", artifact_dir / "service_auth_create_account.json"), car_path=path_from_env("REPO_CAR", artifact_dir / "tempestpds.repo.car"), source_blobs_path=path_from_env("SOURCE_BLOBS_JSON", artifact_dir / "source_blobs.json"), create_account_path=path_from_env("TEMPEST_CREATE_ACCOUNT_JSON", artifact_dir / "tempest_create_account.json"), import_repo_path=path_from_env("TEMPEST_IMPORT_REPO_JSON", artifact_dir / "tempest_import_repo.json"), status_path=path_from_env("TEMPEST_STATUS_JSON", artifact_dir / "tempest_account_status.json"), missing_blobs_path=path_from_env("TEMPEST_MISSING_BLOBS_JSON", artifact_dir / "tempest_missing_blobs.json"), plc_recommended_path=path_from_env("PLC_RECOMMENDED_JSON", artifact_dir / "plc_recommended.json"), plc_token_path=path_from_env("PLC_TOKEN_JSON", artifact_dir / "plc_token.json"), plc_signed_path=path_from_env("PLC_SIGNED_OPERATION_JSON", artifact_dir / "plc_signed_operation.json"), plc_submit_path=path_from_env("PLC_SUBMIT_JSON", artifact_dir / "plc_submit.json"), activate_path=path_from_env("TEMPEST_ACTIVATE_JSON", artifact_dir / "tempest_activate_account.json"), )
def log(message: str) -> None: print(message, flush=True)
def step(name: str) -> None: log(f"\n==> {name}")
def ensure_artifact_dir(settings: Settings) -> None: settings.artifact_dir.mkdir(parents=True, exist_ok=True)
def read_json(path: Path) -> dict[str, Any]: try: with path.open("r", encoding="utf-8") as fh: data = json.load(fh) except FileNotFoundError as exc: raise CliError(f"missing artifact: {path}") from exc except json.JSONDecodeError as exc: raise CliError(f"invalid JSON artifact: {path}") from exc
if not isinstance(data, dict): raise CliError(f"expected JSON object in {path}") return data
def write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as fh: json.dump(data, fh, indent=2, sort_keys=True) fh.write("\n")
def write_bytes(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data)
def safe_summary(data: dict[str, Any]) -> dict[str, Any]: summary: dict[str, Any] = {} for key, value in data.items(): if key.lower() in {"accessjwt", "refreshjwt", "token", "serviceauth", "password"}: summary[f"has_{key}"] = isinstance(value, str) and value != "" elif isinstance(value, (str, int, bool)) or value is None: summary[key] = value return summary
def print_json_summary(label: str, data: dict[str, Any]) -> None: log(f"{label}: {json.dumps(safe_summary(data), sort_keys=True)}")
def request( method: str, url: str, *, headers: dict[str, str] | None = None, json_body: dict[str, Any] | None = None, body: bytes | None = None, timeout: int = 30,) -> tuple[int, dict[str, str], bytes]: headers = dict(headers or {}) payload = body
if json_body is not None: payload = json.dumps(json_body, separators=(",", ":")).encode("utf-8") headers.setdefault("Content-Type", "application/json")
try: with httpx.Client(timeout=timeout, follow_redirects=True) as client: response = client.request(method, url, headers=headers, content=payload) except httpx.HTTPError as exc: raise CliError(f"request failed for {url}: {exc}") from exc
return response.status_code, {key.lower(): value for key, value in response.headers.items()}, response.content
def expect_json(status: int, raw: bytes, url: str) -> dict[str, Any]: try: data = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as exc: raise CliError(f"{url} returned HTTP {status} with non-JSON body") from exc
if not isinstance(data, dict): raise CliError(f"{url} returned HTTP {status} with non-object JSON")
if not 200 <= status <= 299: message = data.get("message") or data.get("error") or raw.decode("utf-8", errors="replace") raise CliError(f"{url} returned HTTP {status}: {message}")
return data
def bearer(token: str) -> dict[str, str]: return {"Authorization": f"Bearer {token}"}
def access_from_session(settings: Settings) -> str: if not settings.old_session_path.exists(): explicit = env("OLD_ACCESS") if explicit: return explicit raise CliError(f"{settings.old_session_path} does not exist; run login-source or set OLD_ACCESS")
data = read_json(settings.old_session_path) session_host = data.get("_tempest_old_login_pds") or data.get("_tempest_old_auth_pds") if isinstance(session_host, str) and session_host != settings.old_login_pds: raise CliError( f"{settings.old_session_path} was created for {session_host}, " f"but OLD_LOGIN_PDS is {settings.old_login_pds}; rerun login-source successfully" )
session_identifier = data.get("_tempest_old_identifier") current_identifier = env("OLD_IDENTIFIER", settings.handle) if isinstance(session_identifier, str) and session_identifier != current_identifier: raise CliError( f"{settings.old_session_path} was created for identifier {session_identifier}, " f"but OLD_IDENTIFIER is {current_identifier}; rerun login-source successfully" )
token = data.get("accessJwt") if not isinstance(token, str) or not token: raise CliError(f"{settings.old_session_path} does not contain accessJwt") return token
def service_auth_token(settings: Settings) -> str: explicit = env("SERVICE_AUTH") if explicit: return explicit
data = read_json(settings.service_auth_path) token = data.get("token") if not isinstance(token, str) or not token: raise CliError(f"{settings.service_auth_path} does not contain token") return token
def tempest_access_token(settings: Settings) -> str: if not settings.create_account_path.exists(): explicit = env("TEMPEST_ACCESS") if explicit: return explicit raise CliError(f"{settings.create_account_path} does not exist; run create-account or set TEMPEST_ACCESS")
data = read_json(settings.create_account_path) token = data.get("accessJwt") if not isinstance(token, str) or not token: raise CliError(f"{settings.create_account_path} does not contain accessJwt") return token
def tempest_refresh_token(settings: Settings) -> str: if not settings.create_account_path.exists(): explicit = env("TEMPEST_REFRESH") if explicit: return explicit raise CliError(f"{settings.create_account_path} does not exist; run create-account or set TEMPEST_REFRESH")
data = read_json(settings.create_account_path) token = data.get("refreshJwt") if not isinstance(token, str) or not token: raise CliError(f"{settings.create_account_path} does not contain refreshJwt") return token
def plc_operation_token(settings: Settings) -> str: explicit = env("PLC_TOKEN") if explicit: return explicit
data = read_json(settings.plc_token_path) token = data.get("token") if not isinstance(token, str) or not token: raise CliError(f"{settings.plc_token_path} does not contain token; set PLC_TOKEN from the emailed code/token if needed") return token
def login_source(settings: Settings) -> None: step("source session") password = require_env(settings, "OLD_PASSWORD", settings.old_password) url = f"{settings.old_login_pds}/xrpc/com.atproto.server.createSession" identifier = env("OLD_IDENTIFIER", settings.handle) payload = {"identifier": identifier, "password": password} auth_factor_token = env("OLD_AUTH_FACTOR_TOKEN") if auth_factor_token: payload["authFactorToken"] = auth_factor_token log( "login context: " f"old_login_pds={settings.old_login_pds} " f"old_auth_pds={settings.old_auth_pds} " f"identifier={identifier} " f"has_old_password={bool(password)} " f"has_old_auth_factor_token={bool(auth_factor_token)}" ) status, _headers, raw = request("POST", url, json_body=payload) data = expect_json(status, raw, url) data["_tempest_old_login_pds"] = settings.old_login_pds data["_tempest_old_auth_pds"] = settings.old_auth_pds data["_tempest_old_identifier"] = identifier write_json(settings.old_session_path, data) print_json_summary("saved source session", data) log(f"wrote {settings.old_session_path}")
def get_service_auth(settings: Settings) -> None: step("service auth") query = urllib.parse.urlencode({"aud": settings.tempest_service_did, "lxm": CREATE_ACCOUNT_LXM}) url = f"{settings.old_pds}/xrpc/com.atproto.server.getServiceAuth?{query}" status, _headers, raw = request("GET", url, headers=bearer(access_from_session(settings))) data = expect_json(status, raw, url) write_json(settings.service_auth_path, data) print_json_summary("saved service auth", data) log(f"aud={settings.tempest_service_did} lxm={CREATE_ACCOUNT_LXM}") log(f"wrote {settings.service_auth_path}")
def source_session_status(settings: Settings) -> None: step("source session status") url = f"{settings.old_auth_pds}/xrpc/com.atproto.server.getSession" status, _headers, raw = request("GET", url, headers=bearer(access_from_session(settings))) data = expect_json(status, raw, url) print_json_summary("source session is accepted", data)
def export_car(settings: Settings) -> None: step("export repo CAR") query = urllib.parse.urlencode({"did": settings.did}) url = f"{settings.old_pds}/xrpc/com.atproto.sync.getRepo?{query}" status, headers, raw = request("GET", url, timeout=60) if not 200 <= status <= 299: data = expect_json(status, raw, url) raise CliError(str(data)) write_bytes(settings.car_path, raw) log(f"content-type={headers.get('content-type', '<unknown>')} bytes={len(raw)}") log(f"wrote {settings.car_path}")
def list_source_blobs(settings: Settings) -> None: step("source blob inventory") query = urllib.parse.urlencode({"did": settings.did}) url = f"{settings.old_pds}/xrpc/com.atproto.sync.listBlobs?{query}" status, _headers, raw = request("GET", url) data = expect_json(status, raw, url) write_json(settings.source_blobs_path, data) cids = data.get("cids") if isinstance(data.get("cids"), list) else [] log(f"blob_count={len(cids)}") log(f"wrote {settings.source_blobs_path}")
def blob_path(settings: Settings, cid: str) -> Path: return settings.artifact_dir / f"tempestpds.blob.{cid}"
def download_source_blobs(settings: Settings) -> None: step("download source blobs") data = read_json(settings.source_blobs_path) cids = data.get("cids") if not isinstance(cids, list): raise CliError(f"{settings.source_blobs_path} does not contain cids")
if not cids: log("no source blobs to download") return
for cid in cids: if not isinstance(cid, str) or not cid: raise CliError(f"invalid blob CID in {settings.source_blobs_path}") query = urllib.parse.urlencode({"did": settings.did, "cid": cid}) url = f"{settings.old_pds}/xrpc/com.atproto.sync.getBlob?{query}" status, headers, raw = request("GET", url, timeout=60) if not 200 <= status <= 299: expect_json(status, raw, url) path = blob_path(settings, cid) write_bytes(path, raw) log(f"downloaded cid={cid} content-type={headers.get('content-type', '<unknown>')} bytes={len(raw)} path={path}")
def create_account(settings: Settings) -> None: step("create inactive Tempest account") email = require_env(settings, "EMAIL", settings.email) password = require_env(settings, "TEMPEST_PASSWORD", settings.tempest_password) payload = { "did": settings.did, "handle": settings.handle, "email": email, "password": password, "serviceAuth": service_auth_token(settings), } url = f"{settings.tempest}/xrpc/com.atproto.server.createAccount" status, _headers, raw = request("POST", url, json_body=payload) data = expect_json(status, raw, url) write_json(settings.create_account_path, data) print_json_summary("saved Tempest account", data) log(f"wrote {settings.create_account_path}")
def refresh_tempest_session(settings: Settings) -> None: step("refresh Tempest session") url = f"{settings.tempest}/xrpc/com.atproto.server.refreshSession" status, _headers, raw = request("POST", url, headers=bearer(tempest_refresh_token(settings))) data = expect_json(status, raw, url) write_json(settings.create_account_path, data) print_json_summary("saved refreshed Tempest account", data) log(f"wrote {settings.create_account_path}")
def import_repo(settings: Settings) -> None: step("import repo CAR into Tempest") if not settings.car_path.exists(): raise CliError(f"missing CAR: {settings.car_path}") url = f"{settings.tempest}/xrpc/com.atproto.repo.importRepo" status, _headers, raw = request( "POST", url, headers={**bearer(tempest_access_token(settings)), "Content-Type": "application/vnd.ipld.car"}, body=settings.car_path.read_bytes(), timeout=120, ) data = expect_json(status, raw, url) write_json(settings.import_repo_path, data) print_json_summary("saved import result", data) log(f"wrote {settings.import_repo_path}")
def check_status(settings: Settings) -> None: step("Tempest account status") url = f"{settings.tempest}/xrpc/com.atproto.server.checkAccountStatus" status, _headers, raw = request("GET", url, headers=bearer(tempest_access_token(settings))) data = expect_json(status, raw, url) write_json(settings.status_path, data) print_json_summary("saved account status", data) log(f"wrote {settings.status_path}")
def list_missing_blobs(settings: Settings) -> None: step("Tempest missing blobs") url = f"{settings.tempest}/xrpc/com.atproto.repo.listMissingBlobs" status, _headers, raw = request("GET", url, headers=bearer(tempest_access_token(settings))) data = expect_json(status, raw, url) write_json(settings.missing_blobs_path, data) blobs = data.get("blobs") if isinstance(data.get("blobs"), list) else [] log(f"missing_blob_count={len(blobs)}") log(f"wrote {settings.missing_blobs_path}")
def upload_missing_blobs(settings: Settings) -> None: step("upload missing blobs to Tempest") data = read_json(settings.missing_blobs_path) blobs = data.get("blobs") if not isinstance(blobs, list): raise CliError(f"{settings.missing_blobs_path} does not contain blobs") if not blobs: log("no missing blobs to upload") return
for blob in blobs: cid = blob.get("cid") if isinstance(blob, dict) else None if not isinstance(cid, str) or not cid: raise CliError(f"invalid missing blob entry in {settings.missing_blobs_path}")
path = blob_path(settings, cid) if not path.exists(): raise CliError(f"missing downloaded blob file: {path}")
mime_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream" if path.name.endswith(f".{cid}"): mime_type = "application/octet-stream" if cid == "bafkreifodmypic3zbjtevk7rbftxvjxgpgegt5njaxn57lamxracv2a3he": mime_type = "image/png"
url = f"{settings.tempest}/xrpc/com.atproto.repo.uploadBlob" status, _headers, raw = request( "POST", url, headers={**bearer(tempest_access_token(settings)), "Content-Type": mime_type}, body=path.read_bytes(), timeout=60, ) result = expect_json(status, raw, url) print_json_summary(f"uploaded cid={cid}", result)
def plc_recommended(settings: Settings) -> None: step("recommended PLC credentials from Tempest") url = f"{settings.tempest}/xrpc/com.atproto.identity.getRecommendedDidCredentials" status, _headers, raw = request("GET", url, headers=bearer(tempest_access_token(settings))) data = expect_json(status, raw, url) write_json(settings.plc_recommended_path, data) print_json_summary("saved recommended PLC credentials", data) log(f"wrote {settings.plc_recommended_path}")
def plc_request_token(settings: Settings) -> None: step("request PLC operation token from old PDS") url = f"{settings.old_auth_pds}/xrpc/com.atproto.identity.requestPlcOperationSignature" status, _headers, raw = request("POST", url, headers=bearer(access_from_session(settings)))
first_error: dict[str, Any] = {} if not 200 <= status <= 299: try: decoded = json.loads(raw.decode("utf-8")) first_error = decoded if isinstance(decoded, dict) else {} except json.JSONDecodeError: first_error = {}
first_message = first_error.get("message") first_error_name = first_error.get("error")
if status == 400 and settings.old_password and first_message == "password is required": status, _headers, raw = request( "POST", url, headers=bearer(access_from_session(settings)), json_body={"password": settings.old_password}, )
if not 200 <= status <= 299: try: decoded = json.loads(raw.decode("utf-8")) data = decoded if isinstance(decoded, dict) else {} except json.JSONDecodeError: data = first_error
message = data.get("message") or first_message error_name = data.get("error") or first_error_name if message == "Bad token scope": raise CliError( "source PDS rejected the fresh session for PLC operation signatures: " f"{error_name or 'InvalidRequest'}: {message}. " "The source session itself is valid, but this old-PDS endpoint is refusing its scope." )
if 200 <= status <= 299 and raw.strip() == b"": data = {"requested": True, "delivery": "email"} else: data = expect_json(status, raw, url)
write_json(settings.plc_token_path, data) print_json_summary("saved PLC operation token response", data) log(f"wrote {settings.plc_token_path}") if "token" not in data: log("Check the source account email for the PLC token/code, then export it as PLC_TOKEN before plc-sign.")
def plc_sign(settings: Settings) -> None: step("sign PLC operation on old PDS") recommended = read_json(settings.plc_recommended_path) payload = { key: recommended[key] for key in ("rotationKeys", "alsoKnownAs", "verificationMethods", "services") if key in recommended } payload["token"] = plc_operation_token(settings)
url = f"{settings.old_auth_pds}/xrpc/com.atproto.identity.signPlcOperation" status, _headers, raw = request("POST", url, headers=bearer(access_from_session(settings)), json_body=payload) data = expect_json(status, raw, url) write_json(settings.plc_signed_path, data) operation = data.get("operation") if isinstance(data.get("operation"), dict) else {} service = operation.get("services", {}).get("atproto_pds", {}) if isinstance(operation.get("services"), dict) else {} log(f"signed operation service_endpoint={service.get('endpoint', '<unknown>')}") log(f"wrote {settings.plc_signed_path}")
def plc_submit(settings: Settings) -> None: step("submit PLC operation through Tempest") data = read_json(settings.plc_signed_path) operation = data.get("operation") if not isinstance(operation, dict): raise CliError(f"{settings.plc_signed_path} does not contain operation")
url = f"{settings.tempest}/xrpc/com.atproto.identity.submitPlcOperation" status, _headers, raw = request("POST", url, headers=bearer(tempest_access_token(settings)), json_body={"operation": operation}) result = expect_json(status, raw, url) write_json(settings.plc_submit_path, result) print_json_summary("saved PLC submit result", result) log(f"wrote {settings.plc_submit_path}")
def activate_account(settings: Settings) -> None: step("activate Tempest account") url = f"{settings.tempest}/xrpc/com.atproto.server.activateAccount" status, _headers, raw = request("POST", url, headers=bearer(tempest_access_token(settings)), json_body={}) data = expect_json(status, raw, url) write_json(settings.activate_path, data) print_json_summary("saved activation result", data) log(f"wrote {settings.activate_path}")
def full(settings: Settings) -> None: started = time.monotonic() log("Tempest migration CLI") log(f"old_pds={settings.old_pds}") log(f"old_login_pds={settings.old_login_pds}") log(f"old_auth_pds={settings.old_auth_pds}") log(f"tempest={settings.tempest}") log(f"tempest_service_did={settings.tempest_service_did}") log(f"did={settings.did}") log(f"handle={settings.handle}") ensure_artifact_dir(settings) login_source(settings) get_service_auth(settings) export_car(settings) list_source_blobs(settings) download_source_blobs(settings) create_account(settings) import_repo(settings) check_status(settings) list_missing_blobs(settings) log(f"\ncomplete in {time.monotonic() - started:.1f}s") log("Do not activate until the DID document #atproto_pds serviceEndpoint points at Tempest.")
def parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="tempest", description="Run Tempest account migration steps from environment variables.", epilog="Admin token helper aliases: tempest ar, tempest arg2, tempest argon.", ) p.add_argument( "command", nargs="?", choices=[command.value for command in Command], default=Command.FULL.value, help="Command to run. Defaults to full.", ) p.add_argument( "--artifact-dir", default=None, help="Directory for JSON/CAR/blob artifacts. Defaults to ARTIFACT_DIR or .sandbox.", ) return p
def run_command(command: Command, settings: Settings) -> None: match command: case Command.FULL: full(settings) case Command.LOGIN_SOURCE: login_source(settings) case Command.SERVICE_AUTH: get_service_auth(settings) case Command.SOURCE_SESSION_STATUS: source_session_status(settings) case Command.EXPORT_CAR: export_car(settings) case Command.LIST_SOURCE_BLOBS: list_source_blobs(settings) case Command.DOWNLOAD_SOURCE_BLOBS: download_source_blobs(settings) case Command.CREATE_ACCOUNT: create_account(settings) case Command.REFRESH_SESSION: refresh_tempest_session(settings) case Command.IMPORT_REPO: import_repo(settings) case Command.STATUS: check_status(settings) case Command.MISSING_BLOBS: list_missing_blobs(settings) case Command.UPLOAD_MISSING_BLOBS: upload_missing_blobs(settings) case Command.PLC_RECOMMENDED: plc_recommended(settings) case Command.PLC_REQUEST_TOKEN: plc_request_token(settings) case Command.PLC_SIGN: plc_sign(settings) case Command.PLC_SUBMIT: plc_submit(settings) case Command.ACTIVATE: activate_account(settings)
def main(argv: list[str] | None = None) -> int: argv = list(sys.argv[1:] if argv is None else argv)
if argv and argv[0] in ARGON_COMMANDS: return ar.main(argv[1:])
args = parser().parse_args(argv)
settings = settings_from_env(args) command = Command(args.command)
try: ensure_artifact_dir(settings) run_command(command, settings) except CliError as exc: log(f"\nERROR: {exc}") return 1 except KeyboardInterrupt: log("\ninterrupted") return 130
return 0
if __name__ == "__main__": sys.exit(main())