#!/usr/bin/env python3 """MoDPR CLI — type `modpr` instead of `docker` for everyday commands. Talks to the MoDPR registry at PUBLIC_BASE_URL / MODPR_URL (https://modpr.modevops.fun in production, http://localhost:8080 locally). Source of truth: cli/modpr.py (scripts/modpr is a copy — keep them in sync). build/push/pull/tag prefix the registry host so tags stay `user/name:tag`. run/ps/images stay user-scoped via label modpr.owner when logged in. Official short names (nginx, redis, postgres, …) stay Docker Hub names. """ from __future__ import annotations import base64 import getpass import json import os import shlex import subprocess import sys from pathlib import Path from typing import Optional, Sequence from urllib.parse import urlparse from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError OWNER_LABEL = "modpr.owner" MANAGED_LABEL = "modpr.managed" DEFAULT_LOCAL_HOST = "localhost:8080" DEFAULT_PROD_URL = "https://modpr.modevops.fun" # Curated Docker Hub official/library short names (keep in sync with # backend/common/modpr_common/opensource.py). OPEN_SOURCE_NAMES = frozenset( { "alpine", "busybox", "debian", "golang", "httpd", "mariadb", "memcached", "mongo", "mysql", "nginx", "node", "php", "postgres", "python", "redis", "ubuntu", "wordpress", } ) HOME_DIR = Path(os.environ.get("MODPR_HOME", Path.home() / ".modpr")) CONFIG_FILE = HOME_DIR / "config" CREDENTIALS_FILE = HOME_DIR / "credentials" TOKEN_FILE = HOME_DIR / "token" DOCKER_CONFIG = Path.home() / ".docker" / "config.json" RUN_LONG_VALUE = frozenset( { "add-host", "annotation", "attach", "blkio-weight", "blkio-weight-device", "cap-add", "cap-drop", "cgroup-parent", "cgroupns", "cidfile", "cpu-period", "cpu-quota", "cpu-rt-period", "cpu-rt-runtime", "cpu-shares", "cpus", "cpuset-cpus", "cpuset-mems", "detach-keys", "device", "device-cgroup-rule", "device-read-bps", "device-read-iops", "device-write-bps", "device-write-iops", "dns", "dns-option", "dns-search", "domainname", "entrypoint", "env", "env-file", "expose", "gpus", "group-add", "health-cmd", "health-interval", "health-retries", "health-start-interval", "health-start-period", "health-timeout", "hostname", "ip", "ip6", "ipc", "isolation", "kernel-memory", "label", "label-file", "link", "link-local-ip", "log-driver", "log-opt", "mac-address", "memory", "memory-reservation", "memory-swap", "memory-swappiness", "mount", "name", "net", "network", "network-alias", "oom-score-adj", "pid", "pids-limit", "platform", "publish", "pull", "restart", "runtime", "security-opt", "shm-size", "stop-signal", "stop-timeout", "storage-opt", "sysctl", "tmpfs", "ulimit", "user", "userns", "uts", "volume", "volume-driver", "volumes-from", "workdir", } ) RUN_SHORT_VALUE = { "a": "attach", "c": "cpu-shares", "e": "env", "h": "hostname", "l": "label", "m": "memory", "p": "publish", "u": "user", "v": "volume", "w": "workdir", } COMMAND_MAP = """ login / logout docker login/logout + API session for PaaS build -t user/n:t docker build -t /user/n:t (always prefix) push / pull docker push/pull with host prefix (pull: official names stay Hub) tag src dst docker tag src /dst run docker run + label modpr.owner=$user; smart image rewrite ps / containers docker ps [--filter label=modpr.owner=$user] (--host = unfiltered) images docker images, marked official vs private (--host = unfiltered) inspect docker inspect logs/start/stop docker logs / start / stop rm / rmi docker rm / rmi (rmi smart-rewrites image names) volume / network docker volume / network (ls and other subcommands) compose docker compose … (passthrough) config print registry host and logged-in user org / project / env Platform tenancy (personal org is created on register) secret Encrypt an environment secret (never printed back) deploy Create a digest-pinned deployment (org project env service image) rollback / promote / scale paas ps | paas logs Platform replicas (docker ps/logs stay container-level) graph / scan Artifact graph and Hub evidence for a digest token Mint a modpr_pat_… API token (shown once) yaml Apply a modpr.yaml project document Unknown commands are passed through to docker. """.strip() SUBCOMMAND_HELP = { "login": "usage: modpr login [username]\nLog in to the website API and the OCI registry (same password).\nRun this command by itself — do not paste build/push after it; Docker would read those lines as the password.\nPassword from the prompt, or MODPR_PASSWORD for scripts.", "logout": "usage: modpr logout\nLog out of the MoDPR registry.\nMaps to: docker logout ", "build": "usage: modpr build -t user/name:tag [PATH]\nPrefix -t/--tag with the registry host. Adds modpr.owner when logged in.\nMaps to: docker build -t /user/name:tag [PATH]", "push": "usage: modpr push user/name:tag\nPrefix the ref with the registry host.\nMaps to: docker push /user/name:tag", "pull": "usage: modpr pull nginx | user/name:tag\nOfficial short names stay Docker Hub; user/name is prefixed.\nMaps to: docker pull nginx or docker pull /user/name:tag", "tag": "usage: modpr tag SOURCE user/name:tag\nQualify the destination tag with the registry host.\nMaps to: docker tag SOURCE /user/name:tag", "run": "usage: modpr run [docker-run-flags] IMAGE [CMD...]\nExample: modpr run -itd -p 8080:80 nginx\nOfficial names stay Hub; user/name is prefixed. Adds --label modpr.owner=$user.\nMaps to: docker run --label modpr.owner=$user …", "ps": "usage: modpr ps [-a|--all] [--host] [docker-ps-flags]\nFilter to label=modpr.owner=$user when logged in. --host shows every container.\nMaps to: docker ps [--filter label=modpr.owner=$user]", "containers": "usage: modpr containers [-a|--all] [--host]\nAlias of: modpr ps", "images": "usage: modpr images [--host] [docker-images-flags]\nList local images and mark SOURCE as official or private.\nMaps to: docker images (filtered to the user unless --host)", "inspect": "usage: modpr inspect [docker-inspect-flags]\nMaps to: docker inspect", "logs": "usage: modpr logs [docker-logs-flags] CONTAINER\nMaps to: docker logs", "stop": "usage: modpr stop [CONTAINER...]\nMaps to: docker stop", "start": "usage: modpr start [CONTAINER...]\nMaps to: docker start", "rm": "usage: modpr rm [docker-rm-flags] CONTAINER...\nMaps to: docker rm", "rmi": "usage: modpr rmi [docker-rmi-flags] IMAGE...\nSmart-rewrites image names, then: docker rmi", "volume": "usage: modpr volume ls | modpr volume …\nMaps to: docker volume", "network": "usage: modpr network ls | modpr network …\nMaps to: docker network", "compose": "usage: modpr compose [docker-compose-args]\nMaps to: docker compose", "config": "usage: modpr config\nPrint registry= and user= for the current environment.", "org": "usage: modpr org ls | modpr org create SLUG\nList organizations or create a team org.", "project": "usage: modpr project ls ORG | modpr project create ORG SLUG\nProjects under an organization.", "env": "usage: modpr env ls ORG PROJECT\nList environments (production, staging, previews).", "secret": "usage: modpr secret set ORG PROJECT ENV NAME VALUE\nStore an encrypted secret. It is never returned after set.", "deploy": "usage: modpr deploy ORG PROJECT ENV SERVICE IMAGE\nResolve the tag to a digest, evaluate policy, run replicas.\nSubcommands: list SERVICE_ID | rollback DEPLOYMENT_ID | promote ORG PROJECT SRC DST SERVICE | scale SERVICE_ID N", "rollback": "usage: modpr rollback DEPLOYMENT_ID\nRe-deploy the digest from that earlier deployment.", "promote": "usage: modpr promote ORG PROJECT SRC DST SERVICE\nPromote the active digest (usually staging → production).", "scale": "usage: modpr scale SERVICE_ID N\nChange desired replicas and roll the same digest.", "paas": "usage: modpr paas ps ORG PROJECT ENV | modpr paas logs DEPLOYMENT_ID\nPlatform process list and deployment logs.", "graph": "usage: modpr graph sha256:…\nArtifact-to-runtime graph for one digest.", "scan": "usage: modpr scan sha256:…\nSBOM / CVE / signature evidence for a digest.", "token": "usage: modpr token mint ORG NAME\nMint a modpr_pat_… token (printed once).\nSame as Team → Mint API token on the website.\nThen: export MODPR_TOKEN=modpr_pat_… and use modpr as usual (paas ps, deploy, …).\nDo not commit the token or paste it into chat.", "yaml": "usage: modpr yaml apply ORG FILE\nApply a modpr.yaml (kind: Project) document.", } def host_from_url(url: str) -> str: """https://modpr.modevops.fun → modpr.modevops.fun; http://localhost:8080/ → localhost:8080.""" raw = (url or "").strip() if not raw: return "" if "://" not in raw: raw = "http://" + raw parsed = urlparse(raw) host = parsed.netloc or parsed.path.split("/")[0] return host.strip().rstrip("/") def _kv_file(path: Path) -> dict[str, str]: data: dict[str, str] = {} if not path.is_file(): return data for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) data[key.strip()] = value.strip().strip("'").strip('"') return data def detect_host( environ: Optional[dict[str, str]] = None, config_file: Optional[Path] = None, credentials_file: Optional[Path] = None, ) -> str: env = os.environ if environ is None else environ if env.get("MODPR_REGISTRY"): return host_from_url(env["MODPR_REGISTRY"]) or env["MODPR_REGISTRY"].strip() for key in ("MODPR_URL", "PUBLIC_BASE_URL"): if env.get(key): host = host_from_url(env[key]) if host: return host creds = _kv_file(credentials_file or CREDENTIALS_FILE) if creds.get("registry"): return host_from_url(creds["registry"]) or creds["registry"] cfg = _kv_file(config_file or CONFIG_FILE) if cfg.get("registry"): return host_from_url(cfg["registry"]) or cfg["registry"] return DEFAULT_LOCAL_HOST def _split_ref(ref: str) -> tuple[str, str, str]: """Return (name, :tag or '', @digest or '').""" raw = (ref or "").strip() digest = "" if "@" in raw: raw, digest = raw.rsplit("@", 1) digest = "@" + digest last_slash = raw.rfind("/") last_colon = raw.rfind(":") tag = "" name = raw if last_colon > last_slash: name, tag = raw[:last_colon], raw[last_colon:] return name, tag, digest def has_explicit_host(name: str) -> bool: first = name.split("/", 1)[0] return "." in first or ":" in first or first == "localhost" def is_official_short(name: str) -> bool: """True for Hub library names: nginx, redis, library/postgres.""" if has_explicit_host(name): return False parts = [p for p in name.split("/") if p] if len(parts) == 1: return True if len(parts) == 2 and parts[0] == "library": return True return False def is_open_source_name(name: str) -> bool: parts = [p for p in name.split("/") if p] if has_explicit_host(name): parts = parts[1:] if parts and parts[0] == "library": parts = parts[1:] return bool(parts) and parts[0].lower() in OPEN_SOURCE_NAMES and len(parts) == 1 def qualify_always(ref: str, host: str) -> str: """Prefix registry host unless the ref already names a host (build/push/tag dest).""" name, tag, digest = _split_ref(ref) if not name: return ref if has_explicit_host(name): return ref return f"{host}/{name}{tag}{digest}" def qualify_image(ref: str, host: str) -> str: """Smart rewrite: official short names stay Hub; user/name is private.""" name, tag, digest = _split_ref(ref) if not name or has_explicit_host(name): return ref if is_official_short(name): return ref return f"{host}/{name}{tag}{digest}" def qualify(ref: str, host: Optional[str] = None) -> str: """Back-compat alias used by older tests/docs: always prefix like build/push.""" return qualify_always(ref, host or detect_host()) def _username_from_docker_auth(host: str) -> Optional[str]: if not DOCKER_CONFIG.is_file(): return None try: payload = json.loads(DOCKER_CONFIG.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None auths = payload.get("auths") or {} keys = (host, f"https://{host}", f"http://{host}", f"https://{host}/v2/", f"http://{host}/v2/") for key in keys: entry = auths.get(key) or {} blob = entry.get("auth") if not blob: continue try: decoded = base64.b64decode(blob).decode("utf-8", errors="replace") except (OSError, ValueError): continue user = decoded.split(":", 1)[0].strip() if user: return user return None def logged_in_user( environ: Optional[dict[str, str]] = None, config_file: Optional[Path] = None, credentials_file: Optional[Path] = None, host: Optional[str] = None, ) -> Optional[str]: env = os.environ if environ is None else environ if env.get("MODPR_USER"): return env["MODPR_USER"].strip() or None creds = _kv_file(credentials_file or CREDENTIALS_FILE) if creds.get("user"): return creds["user"] cfg = _kv_file(config_file or CONFIG_FILE) if cfg.get("user"): return cfg["user"] return _username_from_docker_auth(host or detect_host(environ=env, config_file=config_file)) def _write_identity(host: str, user: str) -> None: HOME_DIR.mkdir(parents=True, exist_ok=True) body = f"registry={host}\nuser={user}\n" CONFIG_FILE.write_text(body, encoding="utf-8") CREDENTIALS_FILE.write_text(body, encoding="utf-8") try: os.chmod(CREDENTIALS_FILE, 0o600) os.chmod(CONFIG_FILE, 0o640) except OSError: pass def _clear_user(host: str) -> None: HOME_DIR.mkdir(parents=True, exist_ok=True) CONFIG_FILE.write_text(f"registry={host}\n", encoding="utf-8") if CREDENTIALS_FILE.exists(): CREDENTIALS_FILE.write_text(f"registry={host}\n", encoding="utf-8") if TOKEN_FILE.exists(): try: TOKEN_FILE.unlink() except OSError: pass def owner_labels(user: Optional[str]) -> list[str]: if not user: return [] return ["--label", f"{OWNER_LABEL}={user}", "--label", f"{MANAGED_LABEL}=true"] def _wants_help(args: Sequence[str], *, allow_short_h: bool) -> bool: if not args: return False if args[0] in ("--help", "help"): return True return allow_short_h and args[0] in ("-h",) def root_help(host: Optional[str] = None) -> str: h = host or detect_host() return ( f"MoDPR — type modpr instead of docker (registry {h})\n\n" f"{COMMAND_MAP}\n\n" f"Registry host from MODPR_URL or PUBLIC_BASE_URL " f"(prod {DEFAULT_PROD_URL} → {host_from_url(DEFAULT_PROD_URL)}; " f"local http://localhost:8080 → {DEFAULT_LOCAL_HOST}).\n" "Never put the host in -t; modpr prefixes it like docker.io on Docker Hub.\n" "Set host with MODPR_URL, PUBLIC_BASE_URL, MODPR_REGISTRY, or ~/.modpr/config." ) def rewrite_build_args(args: Sequence[str], host: str, user: Optional[str] = None) -> list[str]: out: list[str] = [] tagged = False i = 0 while i < len(args): token = args[i] if token in ("-t", "--tag") and i + 1 < len(args): out.extend(["-t", qualify_always(args[i + 1], host)]) tagged = True i += 2 continue if token.startswith("--tag="): out.append("--tag=" + qualify_always(token.split("=", 1)[1], host)) tagged = True i += 1 continue out.append(token) i += 1 if not tagged: raise ValueError("usage: modpr build -t username/name:tag [PATH]") skip = False has_context = False for token in out: if skip: skip = False continue if token in ("-t", "--tag", "-f", "--file"): skip = True continue if token.startswith("--tag=") or token.startswith("--file="): continue if not token.startswith("-"): has_context = True if not has_context: out.append(".") labels = owner_labels(user) if labels and f"{OWNER_LABEL}={user}" not in " ".join(out): out[0:0] = labels return out def rewrite_run_args(args: Sequence[str], host: str, user: Optional[str] = None) -> list[str]: """Rewrite the image token and inject owner labels. Typical flags (-itd -p) pass through.""" out: list[str] = [] i = 0 image_at: Optional[int] = None while i < len(args): token = args[i] if image_at is not None: out.append(token) i += 1 continue if token == "--": i += 1 if i < len(args): image_at = len(out) out.append(qualify_image(args[i], host)) i += 1 out.extend(args[i:]) break if token.startswith("--"): name, eq, value = token[2:].partition("=") out.append(token) if eq: i += 1 continue if name in RUN_LONG_VALUE and i + 1 < len(args): out.append(args[i + 1]) i += 2 continue i += 1 continue if token.startswith("-") and token != "-": cluster = token[1:] if cluster[0] in RUN_SHORT_VALUE and len(cluster) > 1 and not all(c in "ditPq" for c in cluster): # Attached value: -p8080:80 out.append(token) i += 1 continue out.append(token) last = cluster[-1] if last in RUN_SHORT_VALUE and not token.startswith("--") and all( (ch in RUN_SHORT_VALUE or ch in "ditPq") for ch in cluster ): if last in RUN_SHORT_VALUE and i + 1 < len(args) and len(cluster) == 1: out.append(args[i + 1]) i += 2 continue i += 1 continue image_at = len(out) out.append(qualify_image(token, host)) i += 1 if image_at is None: raise ValueError("usage: modpr run [OPTIONS] IMAGE [COMMAND]") labels = owner_labels(user) if labels and f"{OWNER_LABEL}={user}" not in " ".join(out): out[image_at:image_at] = labels return out def rewrite_ref_args(args: Sequence[str], host: str, *, smart: bool) -> list[str]: """Qualify the first positional image ref; keep flags.""" fn = qualify_image if smart else qualify_always out: list[str] = [] done = False skip = False for token in args: if skip: out.append(token) skip = False continue if done or token.startswith("-"): out.append(token) if token in ("--platform", "--from", "-f") or ( token.startswith("--") and token.lstrip("-").split("=")[0] in {"platform", "from"} and "=" not in token ): skip = True continue out.append(fn(token, host)) done = True return out def classify_local_image(repository: str, host: str, user: Optional[str]) -> Optional[str]: """Return 'private', 'official', or None (not relevant).""" raw = (repository or "").strip() if not raw or "" in raw: return None name, _, _ = _split_ref(raw) lower = name.lower() prefixes = [] for item in (host, DEFAULT_LOCAL_HOST, "127.0.0.1:8080"): if item: prefixes.append(item.lower() + "/") for prefix in prefixes: if not lower.startswith(prefix): continue rest = lower[len(prefix) :] ns = rest.split("/", 1)[0] if rest else "" if user and ns == user.lower(): return "private" if "/" in rest: return "private" return None if is_official_short(name) or is_open_source_name(name): return "official" if not has_explicit_host(name) and "/" in name: if user and name.split("/", 1)[0].lower() == user.lower(): return "private" return "private" return None def _fmt_row(cols: Sequence[str], widths: Sequence[int]) -> str: parts = [] for i, col in enumerate(cols): if i == len(cols) - 1: parts.append(col) else: parts.append(col.ljust(widths[i])) return " ".join(parts) def _docker( args: Sequence[str], *, echo: bool = False, replace: bool = False, check: bool = False, ) -> int: cmd = ["docker", *args] if echo: print("docker " + " ".join(shlex.quote(a) for a in args), file=sys.stderr) try: if replace: os.execvp("docker", cmd) completed = subprocess.run(cmd, check=False) if check and completed.returncode != 0: raise subprocess.CalledProcessError(completed.returncode, cmd) return completed.returncode except FileNotFoundError: print("modpr: docker is not installed or not on PATH", file=sys.stderr) return 127 def _strip_host_flag(args: Sequence[str]) -> tuple[bool, list[str]]: show_all = False out: list[str] = [] for token in args: if token == "--host": show_all = True continue out.append(token) return show_all, out def _scheme_for_host(host: str) -> str: h = (host or "").lower() if h.startswith("localhost") or h.startswith("127.0.0.1"): return "http" return "https" def advertised_token_realm(host: str) -> str: """WWW-Authenticate realm from GET /v2/ (empty if unreachable).""" url = f"{_scheme_for_host(host)}://{host}/v2/" try: with urlopen(Request(url, method="GET"), timeout=8) as resp: return resp.headers.get("WWW-Authenticate") or resp.headers.get("www-authenticate") or "" except HTTPError as exc: return exc.headers.get("WWW-Authenticate") or exc.headers.get("www-authenticate") or "" except (URLError, OSError, TimeoutError): return "" def _warn_if_localhost_realm(host: str, www: str) -> None: lower = (www or "").lower() if "localhost" not in lower and "127.0.0.1" not in lower: return if host.lower().startswith("localhost") or host.lower().startswith("127.0.0.1"): return print( "modpr: the Hub advertised a localhost token URL to Docker:\n" f" {www}\n" "Docker then tries to log in on this laptop instead of " f"https://{host}/auth/token.\n" "On the VPS, set PUBLIC_BASE_URL=https://modpr.modevops.fun and recreate registry:\n" " cd /opt/modpr && docker compose --env-file .env " "-f docker-compose.yml -f docker-compose.prod.yml up -d --force-recreate registry\n" "Confirm: curl -sSI https://modpr.modevops.fun/v2/ | grep -i www-authenticate", file=sys.stderr, ) def _read_login_password() -> str: env = os.environ.get("MODPR_PASSWORD") if env is not None: return env try: return getpass.getpass("Password: ") except (EOFError, KeyboardInterrupt) as exc: print("modpr: login cancelled", file=sys.stderr) raise SystemExit(1) from exc def _docker_login(host: str, user: str, password: str) -> int: print(f"docker login {host} --password-stdin", file=sys.stderr) try: proc = subprocess.run( ["docker", "login", host, "-u", user, "--password-stdin"], input=password + "\n", text=True, check=False, ) except FileNotFoundError: print("modpr: docker is not installed or not on PATH", file=sys.stderr) return 127 return proc.returncode def cmd_login(args: list[str]) -> int: host = detect_host() user = args[0] if args else os.environ.get("MODPR_USER") or "" if not user: try: user = input("Username: ").strip() except (EOFError, KeyboardInterrupt): print("modpr: login cancelled", file=sys.stderr) return 1 if not user: print("usage: modpr login [username]", file=sys.stderr) return 2 password = _read_login_password() if not password: print( "modpr: empty password. Run `modpr login` by itself and type the website password.\n" "Do not paste build/push/deploy on the next lines — Docker treats those as the password.", file=sys.stderr, ) return 2 _clear_user(host) try: _login_website(host, user, password) except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") print(f"modpr: website login failed ({exc.code}): {detail}", file=sys.stderr) return 1 except (URLError, RuntimeError, TimeoutError, OSError) as extra: print(f"modpr: cannot complete website login ({extra})", file=sys.stderr) return 1 _write_identity(host, user) print(f"modpr: API session stored for {user}", file=sys.stderr) code = _docker_login(host, user, password) if code != 0: _warn_if_localhost_realm(host, advertised_token_realm(host)) print( "modpr: docker login failed. API session is stored, so `modpr deploy` can work,\n" "but `modpr push` needs a successful docker login.\n" "If this is HTTP localhost, add insecure-registries and restart Docker:\n" " sh scripts/insecure-registry.sh", file=sys.stderr, ) return code return 0 def cmd_logout(_args: list[str]) -> int: host = detect_host() _clear_user(host) print(f"docker logout {host}", file=sys.stderr) return _docker(["logout", host]) def cmd_build(args: list[str]) -> int: try: out = rewrite_build_args(args, detect_host(), logged_in_user()) except ValueError as exc: print(str(exc), file=sys.stderr) return 2 return _docker(["build", *out], echo=True) def cmd_push_pull(verb: str, args: list[str]) -> int: if not args: print(f"usage: modpr {verb} user/name:tag", file=sys.stderr) return 2 smart = verb == "pull" out = rewrite_ref_args(args, detect_host(), smart=smart) return _docker([verb, *out], echo=True) def cmd_tag(args: list[str]) -> int: if len(args) < 2: print("usage: modpr tag SOURCE user/name:tag", file=sys.stderr) return 2 src, dst, *rest = args qualified = qualify_always(dst, detect_host()) code = _docker(["tag", src, qualified, *rest], echo=True) if code == 0: print(f"tagged {qualified}") return code def cmd_run(args: list[str]) -> int: try: out = rewrite_run_args(args, detect_host(), logged_in_user()) except ValueError as exc: print(str(exc), file=sys.stderr) return 2 return _docker(["run", *out], echo=True, replace=True) def cmd_ps(args: list[str]) -> int: show_all, rest = _strip_host_flag(args) user = logged_in_user() extra: list[str] = [] if user and not show_all: extra = ["--filter", f"label={OWNER_LABEL}={user}"] return _docker(["ps", *extra, *rest]) def cmd_images(args: list[str]) -> int: show_all, rest = _strip_host_flag(args) host = detect_host() user = logged_in_user() fmt = "{{.Repository}}\t{{.Tag}}\t{{.ID}}\t{{.Size}}\t{{.CreatedSince}}" cmd = ["docker", "images", f"--format={fmt}", *rest] try: proc = subprocess.run(cmd, check=False, capture_output=True, text=True) except FileNotFoundError: print("modpr: docker is not installed or not on PATH", file=sys.stderr) return 127 if proc.returncode != 0: sys.stderr.write(proc.stderr) return proc.returncode rows = [("SOURCE", "REPOSITORY", "TAG", "IMAGE ID", "SIZE", "CREATED")] for line in proc.stdout.splitlines(): parts = line.split("\t") if len(parts) < 5: continue repo, tag, image_id, size, created = parts[0], parts[1], parts[2], parts[3], parts[4] display = repo if tag in ("", "") else f"{repo}:{tag}" kind = classify_local_image(display if tag != "" else repo, host, user) if kind is None and not show_all: continue source = kind or "other" if not show_all and user and kind == "private": name, _, _ = _split_ref(repo) rest_name = name for prefix in (host + "/", DEFAULT_LOCAL_HOST + "/", "127.0.0.1:8080/"): if rest_name.startswith(prefix): rest_name = rest_name[len(prefix) :] break ns = rest_name.split("/", 1)[0].lower() if rest_name else "" if ns and ns != user.lower(): continue rows.append((source, repo, tag, image_id, size, created)) if len(rows) == 1: print("No images to list. Use --host to show every local docker image.") return 0 widths = [max(len(row[i]) for row in rows) for i in range(5)] for row in rows: print(_fmt_row(row, widths)) return 0 def cmd_rmi(args: list[str]) -> int: out = rewrite_ref_args(args, detect_host(), smart=True) # rewrite_ref_args only qualifies the first positional; qualify every image. done_flags = False final: list[str] = [] skip = False for token in args: if skip: final.append(token) skip = False continue if token.startswith("-"): final.append(token) if token in ("-f",) or (token.startswith("--") and "=" not in token and token.lstrip("-") in {"filter"}): skip = True continue done_flags = True final.append(qualify_image(token, detect_host())) if not done_flags: final = out return _docker(["rmi", *final], echo=True) def cmd_passthrough(docker_cmd: str, args: list[str]) -> int: return _docker([docker_cmd, *args], replace=docker_cmd in {"logs", "attach", "exec"}) def cmd_config(_args: list[str]) -> int: host = detect_host() user = logged_in_user() or "" print(f"registry={host}") if user: print(f"user={user}") if TOKEN_FILE.is_file() or os.environ.get("MODPR_TOKEN"): print("api=session") return 0 def _api_origin(host: Optional[str] = None) -> str: h = host or detect_host() return f"{_scheme_for_host(h)}://{h}" def _docker_user_pass(host: str) -> tuple[str, str]: if not DOCKER_CONFIG.is_file(): return "", "" try: payload = json.loads(DOCKER_CONFIG.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return "", "" auths = payload.get("auths") or {} keys = (host, f"https://{host}", f"http://{host}", f"https://{host}/v2/", f"http://{host}/v2/") for key in keys: blob = (auths.get(key) or {}).get("auth") if not blob: continue try: decoded = base64.b64decode(blob).decode("utf-8", errors="replace") except (OSError, ValueError): continue if ":" not in decoded: continue user, password = decoded.split(":", 1) if user and password: return user, password return "", "" def _write_access_token(token: str) -> None: HOME_DIR.mkdir(parents=True, exist_ok=True) TOKEN_FILE.write_text(token.strip() + "\n", encoding="utf-8") try: os.chmod(TOKEN_FILE, 0o600) except OSError: pass def _login_website(host: str, user: str, password: str) -> str: body = json.dumps({"username": user, "password": password}).encode() req = Request( f"{_api_origin(host)}/auth/login", data=body, method="POST", headers={"Content-Type": "application/json", "Accept": "application/json"}, ) with urlopen(req, timeout=20) as resp: data = json.loads(resp.read().decode()) token = str(data.get("access_token") or "") if not token: raise RuntimeError("website login returned no access_token") _write_access_token(token) return token def _session_username(token: str) -> str: if not token or token.startswith("modpr_pat_"): return "" parts = token.split(".") if len(parts) < 2: return "" pad = "=" * (-len(parts[1]) % 4) try: payload = json.loads(base64.urlsafe_b64decode(parts[1] + pad).decode("utf-8")) except (OSError, ValueError, json.JSONDecodeError): return "" return str(payload.get("username") or "") def _sync_website_token(host: Optional[str] = None) -> None: h = host or detect_host() user, password = _docker_user_pass(h) if not (user and password): return try: _login_website(h, user, password) except Exception as exc: print(f"modpr: website session not stored ({exc})", file=sys.stderr) def _bearer() -> str: env = (os.environ.get("MODPR_TOKEN") or "").strip() if env: return env if TOKEN_FILE.is_file(): token = TOKEN_FILE.read_text(encoding="utf-8").strip() if token: return token _sync_website_token() if TOKEN_FILE.is_file(): return TOKEN_FILE.read_text(encoding="utf-8").strip() print("modpr: no API session. Run `modpr login` or set MODPR_TOKEN.", file=sys.stderr) raise SystemExit(2) def api_json(method: str, path: str, payload: object | None = None, extra_headers: Optional[dict[str, str]] = None) -> object: token = _bearer() data = None if payload is None else json.dumps(payload).encode() headers = { "Authorization": f"Bearer {token}", "Accept": "application/json", } if data is not None: headers["Content-Type"] = "application/json" if extra_headers: headers.update(extra_headers) url = _api_origin() + path req = Request(url, data=data, method=method.upper(), headers=headers) def _do(): with urlopen(req, timeout=90) as resp: raw = resp.read().decode("utf-8", errors="replace") return json.loads(raw) if raw else {} try: return _do() except HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") if exc.code == 401: _sync_website_token() token2 = (os.environ.get("MODPR_TOKEN") or "").strip() if not token2 and TOKEN_FILE.is_file(): token2 = TOKEN_FILE.read_text(encoding="utf-8").strip() if token2 and token2 != token: headers["Authorization"] = f"Bearer {token2}" req2 = Request(url, data=data, method=method.upper(), headers=headers) try: with urlopen(req2, timeout=90) as resp: raw = resp.read().decode("utf-8", errors="replace") return json.loads(raw) if raw else {} except HTTPError as exc2: body = exc2.read().decode("utf-8", errors="replace") print(f"modpr: API {exc2.code} {path}: {body}", file=sys.stderr) raise SystemExit(1) from exc2 print(f"modpr: API {exc.code} {path}: {body}", file=sys.stderr) who = _session_username(token) if exc.code == 403 and who: print( f"modpr: API session is {who}. Deploy to that user's org " f"(usually `{who} default staging …`), or run: modpr login {who}", file=sys.stderr, ) raise SystemExit(1) from exc except URLError as exc: print(f"modpr: cannot reach {_api_origin()} ({exc})", file=sys.stderr) raise SystemExit(1) from exc def _print_json(data: object) -> int: print(json.dumps(data, indent=2, default=str)) return 0 def cmd_org(args: list[str]) -> int: if not args or args[0] in ("ls", "list"): return _print_json(api_json("GET", "/api/v1/orgs")) if args[0] == "create" and len(args) >= 2: slug = args[1] return _print_json(api_json("POST", "/api/v1/orgs", {"slug": slug, "display_name": slug})) print(SUBCOMMAND_HELP["org"], file=sys.stderr) return 2 def cmd_project(args: list[str]) -> int: if not args: print(SUBCOMMAND_HELP["project"], file=sys.stderr) return 2 if args[0] in ("ls", "list") and len(args) >= 2: return _print_json(api_json("GET", f"/api/v1/orgs/{args[1]}/projects")) if args[0] == "create" and len(args) >= 3: org, slug = args[1], args[2] return _print_json( api_json("POST", f"/api/v1/orgs/{org}/projects", {"slug": slug, "display_name": slug}) ) print(SUBCOMMAND_HELP["project"], file=sys.stderr) return 2 def cmd_env(args: list[str]) -> int: if len(args) >= 3 and args[0] in ("ls", "list"): return _print_json(api_json("GET", f"/api/v1/orgs/{args[1]}/projects/{args[2]}/environments")) print(SUBCOMMAND_HELP["env"], file=sys.stderr) return 2 def cmd_secret(args: list[str]) -> int: if args[:1] == ["set"] and len(args) >= 6: org, project, env, name, value = args[1], args[2], args[3], args[4], args[5] return _print_json( api_json( "POST", f"/api/v1/orgs/{org}/projects/{project}/environments/{env}/secrets", {"name": name, "value": value}, ) ) print(SUBCOMMAND_HELP["secret"], file=sys.stderr) return 2 def _ensure_service(org: str, project: str, env: str, service: str, image: str) -> str: rows = api_json("GET", f"/api/v1/orgs/{org}/projects/{project}/environments/{env}/services") if isinstance(rows, list): for row in rows: if isinstance(row, dict) and row.get("name") == service: return str(row["id"]) created = api_json( "POST", f"/api/v1/orgs/{org}/projects/{project}/environments/{env}/services", {"name": service, "image": image, "service_type": "web", "port": 80}, ) if not isinstance(created, dict): raise SystemExit("modpr: could not create service") return str(created["id"]) def _deploy_create(org: str, project: str, env: str, service: str, image: str) -> int: sid = _ensure_service(org, project, env, service, image) dep = api_json( "POST", f"/api/v1/services/{sid}/deployments", {"image": image}, extra_headers={"Idempotency-Key": f"cli-{os.getpid()}-{service}"}, ) return _print_json(dep) def cmd_deploy(args: list[str]) -> int: if not args: print(SUBCOMMAND_HELP["deploy"], file=sys.stderr) return 2 if args[0] == "list" and len(args) >= 2: return _print_json(api_json("GET", f"/api/v1/services/{args[1]}/deployments")) if args[0] == "logs" and len(args) >= 2: return _print_json(api_json("GET", f"/api/v1/deployments/{args[1]}/logs")) if args[0] == "rollback" and len(args) >= 2: return cmd_rollback(args[1:]) if args[0] == "promote": return cmd_promote(args[1:]) if args[0] == "scale": return cmd_scale(args[1:]) if args[0] == "create" and len(args) >= 6: return _deploy_create(*args[1:6]) if len(args) >= 5 and args[0] not in ("list", "logs", "rollback", "promote", "scale", "create"): return _deploy_create(*args[:5]) print(SUBCOMMAND_HELP["deploy"], file=sys.stderr) return 2 def cmd_rollback(args: list[str]) -> int: if not args: print(SUBCOMMAND_HELP["rollback"], file=sys.stderr) return 2 return _print_json( api_json( "POST", f"/api/v1/deployments/{args[0]}/rollback", {}, extra_headers={"Idempotency-Key": f"cli-rb-{os.getpid()}"}, ) ) def cmd_promote(args: list[str]) -> int: if len(args) < 5: print(SUBCOMMAND_HELP["promote"], file=sys.stderr) return 2 org, project, src, dst, service = args[:5] return _print_json( api_json( "POST", f"/api/v1/orgs/{org}/projects/{project}/environments/{src}/promote/{dst}", {"service": service}, ) ) def cmd_scale(args: list[str]) -> int: if len(args) < 2: print(SUBCOMMAND_HELP["scale"], file=sys.stderr) return 2 return _print_json(api_json("POST", f"/api/v1/services/{args[0]}/scale", {"replicas": int(args[1])})) def cmd_paas(args: list[str]) -> int: if len(args) >= 4 and args[0] == "ps": org, project, env = args[1], args[2], args[3] svcs = api_json("GET", f"/api/v1/orgs/{org}/projects/{project}/environments/{env}/services") out = [] if isinstance(svcs, list): for svc in svcs: if not isinstance(svc, dict): continue deps = api_json("GET", f"/api/v1/services/{svc['id']}/deployments") active = None if isinstance(deps, list) and deps: active = next((d for d in deps if isinstance(d, dict) and d.get("status") == "active"), deps[0]) out.append({"service": svc.get("name"), "id": svc.get("id"), "deployment": active}) return _print_json(out) if len(args) >= 2 and args[0] == "logs": return _print_json(api_json("GET", f"/api/v1/deployments/{args[1]}/logs")) print(SUBCOMMAND_HELP["paas"], file=sys.stderr) return 2 def cmd_graph(args: list[str]) -> int: if not args: print(SUBCOMMAND_HELP["graph"], file=sys.stderr) return 2 digest = args[0] if args[0].startswith("sha256:") else f"sha256:{args[0]}" return _print_json(api_json("GET", f"/api/v1/artifacts/{digest}/graph")) def cmd_scan(args: list[str]) -> int: if not args: print(SUBCOMMAND_HELP["scan"], file=sys.stderr) return 2 digest = args[0] if args[0].startswith("sha256:") else f"sha256:{args[0]}" return _print_json(api_json("GET", f"/catalog/artifacts/{digest}/evidence")) def cmd_token(args: list[str]) -> int: if args[:1] == ["mint"] and len(args) >= 3: org, name = args[1], args[2] data = api_json("POST", f"/api/v1/orgs/{org}/tokens", {"name": name, "scopes": ["deploy", "registry:pull"]}) return _print_json(data) print(SUBCOMMAND_HELP["token"], file=sys.stderr) return 2 def cmd_yaml(args: list[str]) -> int: if args[:1] == ["apply"] and len(args) >= 3: org, path = args[1], args[2] document = Path(path).read_text(encoding="utf-8") return _print_json(api_json("POST", f"/api/v1/orgs/{org}/yaml", {"document": document})) print(SUBCOMMAND_HELP["yaml"], file=sys.stderr) return 2 def main(argv: Optional[Sequence[str]] = None) -> int: argv = list(sys.argv if argv is None else argv) if argv and argv[0].endswith((".py", "modpr")): argv = argv[1:] if not argv or argv[0] in ("-h", "--help", "help"): print(root_help()) return 0 cmd, rest = argv[0], argv[1:] allow_h = cmd != "run" if _wants_help(rest, allow_short_h=allow_h): print(SUBCOMMAND_HELP.get(cmd, root_help())) return 0 dispatch = { "login": cmd_login, "logout": cmd_logout, "build": cmd_build, "push": lambda a: cmd_push_pull("push", a), "pull": lambda a: cmd_push_pull("pull", a), "tag": cmd_tag, "run": cmd_run, "ps": cmd_ps, "containers": cmd_ps, "images": cmd_images, "inspect": lambda a: cmd_passthrough("inspect", a), "logs": lambda a: cmd_passthrough("logs", a), "stop": lambda a: cmd_passthrough("stop", a), "start": lambda a: cmd_passthrough("start", a), "rm": lambda a: cmd_passthrough("rm", a), "rmi": cmd_rmi, "volume": lambda a: cmd_passthrough("volume", a), "network": lambda a: cmd_passthrough("network", a), "compose": lambda a: _docker(["compose", *a], replace=True), "config": cmd_config, "org": cmd_org, "project": cmd_project, "env": cmd_env, "secret": cmd_secret, "deploy": cmd_deploy, "rollback": cmd_rollback, "promote": cmd_promote, "scale": cmd_scale, "paas": cmd_paas, "graph": cmd_graph, "scan": cmd_scan, "token": cmd_token, "yaml": cmd_yaml, } handler = dispatch.get(cmd) if handler is None: return _docker([cmd, *rest], replace=True) return handler(rest) if __name__ == "__main__": raise SystemExit(main(sys.argv))