From c8045b8bff659cc58ffca33af855645693e25b51 Mon Sep 17 00:00:00 2001 From: Sneha Edula Date: Thu, 26 Mar 2026 13:06:32 -0400 Subject: [PATCH 1/3] Intial --- .gitignore | 29 ++++++++++++ README.md | 126 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+) create mode 100644 .gitignore create mode 100644 README.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1708e32 --- /dev/null +++ b/.gitignore @@ -0,0 +1,29 @@ +# Virtual environment +.venv/ +venv/ +env/ + +# Python +__pycache__/ +*.pyc +*.pyo +*.pyd +*.egg-info/ +dist/ +build/ + +# VS Code +.vscode/ +!.vscode/settings.json + +# PyInstaller +*.spec +dist/ +build/ + +# CLI config (user-specific) +.ce/ + +# Misc +.env +*.log \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..0978c43 --- /dev/null +++ b/README.md @@ -0,0 +1,126 @@ +# Credential Engine CLI (`ce`) + +A command-line tool for managing Credential Engine platform resources. + + + +## Requirements + +- Python 3.10+ + + + +## Installation + +```bash +pip install -e . +``` + +This puts the `ce` command on your `$PATH`. + + +## Quick start + +```bash +# Sign in +ce login + +``` + + +## Command reference + +### Authentication + +| Command | Description | +|---|---| +| `ce login` | Sign in via Keycloak device authorization | +| `ce logout` | Sign out and revoke tokens | +| `ce whoami` | Show who you're signed in as | + +### Environments (`ce env`) + +Environments let you switch between `dev`, `sandbox`, and `prod` without touching env vars. Each environment maps a friendly name to an API URL (plus optional OIDC overrides). + +| Command | Description | +|---|---| +| `ce env list` | List all environments (● marks the active one) | +| `ce env use ` | Switch to a different environment | + +```bash +ce env list +ce env use dev +``` + + +### Bulk DID Upload (`ce iir bulk-upload-dids`) + +Register DIDs for multiple organizations in one go. The process has two phases: + +**Phase 1 — Generate signed challenges:** + +```bash +ce iir bulk-upload-dids sign-challenges --csv input.csv +``` + +Validates each row (membership, registry lookup, DID resolution), creates challenges, signs JWTs with the provided private keys, and writes the results to an output CSV. + +**Phase 2 — Publish to the IIR registry:** + +```bash +ce iir bulk-upload-dids publish --output-file Output-input-20240101T000000Z.csv +``` + +Verifies the JWT signatures server-side and publishes each issuer to the IIR. + +#### Input CSV format + +| Column | Required | Description | +|---|---|---| +| `CTID` | Yes | The organization's CTID (must be published in the registry) | +| `DID` | Yes | `did:key:...` or `did:web:...` | +| `VerificationMethod` | Yes | Full verification method ID (e.g., `did:key:z6Mk...#z6Mk...`) | +| `Algorithm` | For did:key | `Ed25519`, `secp256k1`, `P-256`, or `X25519` | +| `PrivateKey` | Yes | Multibase-encoded private key for signing | +| `ValidFrom` | No | Date when the issuer becomes valid (MM/DD/YYYY) | +| `ValidUntil` | No | Date when the issuer expires (MM/DD/YYYY) | + +Example: + +```csv +CTID,DID,VerificationMethod,Algorithm,PrivateKey,ValidFrom,ValidUntil +ce-12345678-...,did:key:z6MkhaX...,did:key:z6MkhaX...#z6MkhaX...,Ed25519,z3u2en...,01/01/2024,12/31/2025 +ce-87654321-...,did:web:example.com,did:web:example.com#key-1,,z3u2en...,, +``` + +`ValidFrom` and `ValidUntil` are optional. + + + + +## Project structure + +``` +ce-cli/ +├── pyproject.toml +└── ce/ + ├── main.py # Root CLI group + top-level aliases + ├── auth/ + │ ├── device_flow.py # RFC 8628 Keycloak client + │ └── token_manager.py # Silent token refresh + require_login() + ├── commands/ + │ ├── auth.py # ce login / logout / account show + │ ├── env.py # ce env list/add/use/show/remove + │ ├── config.py # ce config set/get/list/reset + │ └── resource.py # ce resource list/show/create/delete + ├── config/ + │ ├── settings.py # OIDCSettings, token I/O, paths + │ └── context.py # Environment model, active-env helpers + ├── iir/ + │ ├── csv_processor.py # Bulk DID upload (sign-challenges + publish) + │ └── did_ops.py # DID validation, challenges, JWT signing + └── utils/ + ├── http.py # Authenticated httpx wrapper + APIError + ├── output.py # table/json/yaml/tsv renderer + @output_option + └── errors.py # @handle_api_errors decorator +``` From e94d3b65df43a9e58a325f81ad9efe4270c7d7d2 Mon Sep 17 00:00:00 2001 From: Sneha Edula Date: Thu, 26 Mar 2026 14:02:37 -0400 Subject: [PATCH 2/3] IIR --- ce/__init__.py | 1 + ce/auth/__init__.py | 10 + ce/auth/device_flow.py | 186 +++++++++++++++ ce/auth/token_manager.py | 52 +++++ ce/commands/__init__.py | 1 + ce/commands/auth.py | 215 ++++++++++++++++++ ce/commands/config.py | 136 +++++++++++ ce/commands/env.py | 190 ++++++++++++++++ ce/commands/iir.py | 167 ++++++++++++++ ce/commands/version.py | 59 +++++ ce/config/__init__.py | 29 +++ ce/config/context.py | 178 +++++++++++++++ ce/config/settings.py | 119 ++++++++++ ce/iir/__init__.py | 0 ce/iir/csv_processor.py | 367 ++++++++++++++++++++++++++++++ ce/iir/did_ops.py | 477 +++++++++++++++++++++++++++++++++++++++ ce/main.py | 100 ++++++++ ce/utils/__init__.py | 15 ++ ce/utils/errors.py | 46 ++++ ce/utils/http.py | 106 +++++++++ ce/utils/logging.py | 78 +++++++ ce/utils/output.py | 134 +++++++++++ pyproject.toml | 30 +++ 23 files changed, 2696 insertions(+) create mode 100644 ce/__init__.py create mode 100644 ce/auth/__init__.py create mode 100644 ce/auth/device_flow.py create mode 100644 ce/auth/token_manager.py create mode 100644 ce/commands/__init__.py create mode 100644 ce/commands/auth.py create mode 100644 ce/commands/config.py create mode 100644 ce/commands/env.py create mode 100644 ce/commands/iir.py create mode 100644 ce/commands/version.py create mode 100644 ce/config/__init__.py create mode 100644 ce/config/context.py create mode 100644 ce/config/settings.py create mode 100644 ce/iir/__init__.py create mode 100644 ce/iir/csv_processor.py create mode 100644 ce/iir/did_ops.py create mode 100644 ce/main.py create mode 100644 ce/utils/__init__.py create mode 100644 ce/utils/errors.py create mode 100644 ce/utils/http.py create mode 100644 ce/utils/logging.py create mode 100644 ce/utils/output.py create mode 100644 pyproject.toml diff --git a/ce/__init__.py b/ce/__init__.py new file mode 100644 index 0000000..a68927d --- /dev/null +++ b/ce/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" \ No newline at end of file diff --git a/ce/auth/__init__.py b/ce/auth/__init__.py new file mode 100644 index 0000000..0d593bb --- /dev/null +++ b/ce/auth/__init__.py @@ -0,0 +1,10 @@ +from ce.auth.device_flow import ( + AccessDeniedError, + DeviceAuthResponse, + DeviceFlowError, + KeycloakDeviceFlowClient, + TokenExpiredError, + TokenResponse, +) +from ce.auth.token_manager import get_valid_access_token, is_logged_in, require_login + diff --git a/ce/auth/device_flow.py b/ce/auth/device_flow.py new file mode 100644 index 0000000..726c45b --- /dev/null +++ b/ce/auth/device_flow.py @@ -0,0 +1,186 @@ +"""Keycloak Device Authorization Grant flow""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Optional + +import httpx + +@dataclass +class DeviceAuthResponse: + device_code: str + user_code: str + verification_uri: str + verification_uri_complete: Optional[str] + expires_in: int + interval: int + + +@dataclass +class TokenResponse: + access_token: str + refresh_token: Optional[str] + id_token: Optional[str] + token_type: str + expires_in: int + scope: Optional[str] + + +class DeviceFlowError(Exception): + """Raised when the device flow cannot complete.""" + +class AuthorizationPendingError(DeviceFlowError): + """User has not yet approved, keep polling.""" + +class SlowDownError(DeviceFlowError): + """Server requested a longer polling interval.""" + +class AccessDeniedError(DeviceFlowError): + """User denied the authorization request.""" + +class TokenExpiredError(DeviceFlowError): + """Device code expired before the user approved.""" + +class KeycloakDeviceFlowClient: + """Keycloak device authorization grant.""" + + def __init__(self, env: "ce.config.context.Environment") -> None: + self._env = env + self._http = httpx.Client( + timeout=30, + verify=env.ssl_verify, + ) + + def request_device_code(self) -> DeviceAuthResponse: + resp = self._http.post( + self._env.device_auth_url, + data={ + "client_id": self._env.client_id, + "scope": self._env.scopes, + }, + ) + self._raise_for_status(resp) + body = resp.json() + return DeviceAuthResponse( + device_code=body["device_code"], + user_code=body["user_code"], + verification_uri=body["verification_uri"], + verification_uri_complete=body.get("verification_uri_complete"), + expires_in=body.get("expires_in", 600), + interval=body.get("interval", 5), + ) + + def poll_for_token( + self, + device_code: str, + interval: int, + expires_in: int, + on_pending: Optional[callable] = None, + ) -> TokenResponse: + deadline = time.monotonic() + expires_in + poll_interval = interval + + while time.monotonic() < deadline: + time.sleep(poll_interval) + resp = self._http.post( + self._env.token_url, + data={ + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + "device_code": device_code, + "client_id": self._env.client_id, + }, + ) + body = resp.json() + + if resp.is_success: + return TokenResponse( + access_token=body["access_token"], + refresh_token=body.get("refresh_token"), + id_token=body.get("id_token"), + token_type=body.get("token_type", "Bearer"), + expires_in=body.get("expires_in", 300), + scope=body.get("scope"), + ) + + error = body.get("error", "unknown_error") + + if error == "authorization_pending": + if on_pending: + on_pending() + continue + if error == "slow_down": + poll_interval += 5 + if on_pending: + on_pending() + continue + if error == "access_denied": + raise AccessDeniedError("Authorization was denied by the user.") + if error in ("expired_token", "device_code_expired"): + raise TokenExpiredError("The device code has expired. Please try again.") + raise DeviceFlowError(f"Unexpected error from token endpoint: {error}") + + raise TokenExpiredError("Timed out waiting for user authorization.") + + def refresh_access_token(self, refresh_token: str) -> TokenResponse: + resp = self._http.post( + self._env.token_url, + data={ + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": self._env.client_id, + }, + ) + self._raise_for_status(resp) + body = resp.json() + return TokenResponse( + access_token=body["access_token"], + refresh_token=body.get("refresh_token", refresh_token), + id_token=body.get("id_token"), + token_type=body.get("token_type", "Bearer"), + expires_in=body.get("expires_in", 300), + scope=body.get("scope"), + ) + + def get_userinfo(self, access_token: str) -> dict: + base = self._env.token_url.replace("/token", "") + userinfo_url = f"{base}/userinfo" + resp = self._http.get( + userinfo_url, + headers={"Authorization": f"Bearer {access_token}"}, + ) + self._raise_for_status(resp) + return resp.json() + + def revoke_token(self, token: str, token_type_hint: str = "refresh_token") -> None: + logout_url = self._env.token_url.replace("/token", "/logout") + self._http.post( + logout_url, + data={ + "client_id": self._env.client_id, + "token": token, + "token_type_hint": token_type_hint, + }, + ) + + @staticmethod + def _raise_for_status(resp: httpx.Response) -> None: + if resp.is_error: + try: + detail = resp.json().get("error_description") or resp.text + except Exception: + detail = resp.text + raise DeviceFlowError(f"HTTP {resp.status_code} from {resp.url}: {detail}") + + @staticmethod + def to_stored_tokens(tr: TokenResponse) -> "ce.config.settings.StoredTokens": + from ce.config.settings import StoredTokens + return StoredTokens( + access_token=tr.access_token, + refresh_token=tr.refresh_token, + id_token=tr.id_token, + token_type=tr.token_type, + expires_at=time.time() + tr.expires_in, + scope=tr.scope, + ) \ No newline at end of file diff --git a/ce/auth/token_manager.py b/ce/auth/token_manager.py new file mode 100644 index 0000000..b31483b --- /dev/null +++ b/ce/auth/token_manager.py @@ -0,0 +1,52 @@ +"""Token management, refresh, expiry checks.""" + +from __future__ import annotations + +import time +from typing import Optional + +_REFRESH_BUFFER = 60 # refresh if token expires + + +def get_valid_access_token() -> Optional[str]: + """Return a valid access token, refreshing silently if needed.""" + from ce.config.settings import load_tokens, save_tokens + from ce.config.context import get_current_environment + from ce.auth.device_flow import KeycloakDeviceFlowClient + + tokens = load_tokens() + if tokens is None: + return None + + if tokens.expires_at and time.time() < tokens.expires_at - _REFRESH_BUFFER: + return tokens.access_token + + if tokens.refresh_token: + result = get_current_environment() + if result is None: + return None + _, env = result + client = KeycloakDeviceFlowClient(env) + try: + tr = client.refresh_access_token(tokens.refresh_token) + new_tokens = KeycloakDeviceFlowClient.to_stored_tokens(tr) + save_tokens(new_tokens) + return new_tokens.access_token + except Exception: + return None + + return None + + +def is_logged_in() -> bool: + return get_valid_access_token() is not None + + +def require_login() -> str: + import click + token = get_valid_access_token() + if not token: + raise click.UsageError( + "You are not logged in. Run 'ce login' to authenticate." + ) + return token \ No newline at end of file diff --git a/ce/commands/__init__.py b/ce/commands/__init__.py new file mode 100644 index 0000000..7a5d362 --- /dev/null +++ b/ce/commands/__init__.py @@ -0,0 +1 @@ +"""CE CLI command modules.""" diff --git a/ce/commands/auth.py b/ce/commands/auth.py new file mode 100644 index 0000000..b85bc12 --- /dev/null +++ b/ce/commands/auth.py @@ -0,0 +1,215 @@ +"""Auth command group: ce login, ce logout, ce account show.""" + +from __future__ import annotations + +import click +from rich.console import Console +from rich.panel import Panel +from rich.table import Table + +from ce.auth.device_flow import ( + AccessDeniedError, + DeviceFlowError, + KeycloakDeviceFlowClient, + TokenExpiredError, +) +from ce.auth.token_manager import is_logged_in +from ce.config.context import get_current_environment, load_cli_config, require_environment +from ce.config.settings import delete_tokens, load_tokens, save_tokens + +console = Console() +err_console = Console(stderr=True) + + +@click.group("account") +def auth_group() -> None: + """Manage Credential Engine account and authentication.""" + + +# ce login + +@click.command("login") +@click.option("--env", "env_name", default=None, metavar="NAME", + help="Environment to log in to (default: active environment).") +def login(env_name: str | None) -> None: + """Sign in to Credential Engine via Keycloak device authorization. + + Uses the active environment's device_auth_url, token_url, and client_id. + Override with --env to target a specific environment. + """ + config = load_cli_config() + + if env_name: + if env_name not in config.environments: + err_console.print(f"Unknown environment [bold]{env_name}[/bold]. " + f"Run [bold]ce env list[/bold] to see available environments.") + raise SystemExit(1) + env = config.environments[env_name] + active_name = env_name + else: + result = get_current_environment() + if result is None: + err_console.print("No active environment. Run [bold]ce env use [/bold] first.") + raise SystemExit(1) + active_name, env = result + + if is_logged_in(): + console.print(f"Already signed in to [bold]{active_name}[/bold] " + f"([dim]{env.label}[/dim]). Run [bold]ce account show[/bold] for details.") + console.print(" To switch accounts, run [bold]ce logout[/bold] first.") + return + + client = KeycloakDeviceFlowClient(env) + + console.print() + with console.status(f"[bold]Contacting {env.label} Keycloak...[/bold]"): + try: + device_auth = client.request_device_code() + except DeviceFlowError as exc: + err_console.print(f"Error: {exc}") + raise SystemExit(1) + + panel_content = ( + f"[dim]Environment:[/dim] [bold]{active_name}[/bold] ({env.label})\n\n" + f"[dim]Open a browser and navigate to:[/dim]\n\n" + f" [bold underline]{device_auth.verification_uri}[/bold underline]\n\n" + f"[dim]Then enter the code:[/dim]\n\n" + f" [bold cyan]{device_auth.user_code}[/bold cyan]\n" + ) + if device_auth.verification_uri_complete: + panel_content += ( + f"\n[dim]Or visit the direct link:[/dim]\n" + f" [link={device_auth.verification_uri_complete}]" + f"{device_auth.verification_uri_complete}[/link]\n" + ) + + console.print(Panel( + panel_content, + title="[bold]Credential Engine - Device Sign-In[/bold]", + border_style="blue", + padding=(1, 2), + )) + + try: + with console.status("[dim]Waiting for you to approve in the browser...[/dim]", spinner="dots"): + token_resp = client.poll_for_token( + device_code=device_auth.device_code, + interval=device_auth.interval, + expires_in=device_auth.expires_in, + ) + except AccessDeniedError: + console.print() + err_console.print("Authorization was denied.") + raise SystemExit(1) + except TokenExpiredError: + console.print() + err_console.print("The device code expired. Run [bold]ce login[/bold] again.") + raise SystemExit(1) + except DeviceFlowError as exc: + console.print() + err_console.print(f"Error: {exc}") + raise SystemExit(1) + + stored = KeycloakDeviceFlowClient.to_stored_tokens(token_resp) + save_tokens(stored) + + try: + userinfo = client.get_userinfo(stored.access_token) + name = userinfo.get("name") or userinfo.get("preferred_username") or "there" + email = userinfo.get("email", "") + except Exception: + name, email = "there", "" + + console.print() + greeting = f"Signed in as [bold]{name}[/bold]" + if email: + greeting += f" [dim]({email})[/dim]" + greeting += f" [dim]-> {active_name} ({env.label})[/dim]" + console.print(greeting) + console.print(" Run [bold]ce whoami show[/bold] to view your account details.") + console.print() + + +# ce logout + +@click.command("logout") +@click.option("--yes", "-y", is_flag=True, help="Skip confirmation prompt.") +def logout(yes: bool) -> None: + """Sign out and remove stored credentials.""" + tokens = load_tokens() + if tokens is None: + console.print("You are not currently signed in.") + return + + if not yes: + click.confirm(" Sign out of Credential Engine?", abort=True) + + try: + result = get_current_environment() + if result and tokens.refresh_token: + _, env = result + client = KeycloakDeviceFlowClient(env) + client.revoke_token(tokens.refresh_token, "refresh_token") + except Exception: + pass + + delete_tokens() + console.print("Signed out successfully.") + + +# ce whoami + +@click.command("show") +def account_show() -> None: + """Show the currently signed-in account.""" + from ce.auth.token_manager import get_valid_access_token + + tokens = load_tokens() + if tokens is None: + console.print("Not signed in. Run [bold]ce login[/bold].") + return + + result = get_current_environment() + if result is None: + err_console.print("No active environment.") + return + active_name, env = result + + token = get_valid_access_token() + if not token: + err_console.print("Session expired. Run [bold]ce login[/bold] again.") + return + + client = KeycloakDeviceFlowClient(env) + try: + userinfo = client.get_userinfo(token) + except Exception as exc: + err_console.print(f"Error fetching account info: {exc}") + return + + table = Table(show_header=False, box=None, padding=(0, 2)) + table.add_column("Key", style="dim") + table.add_column("Value", style="bold") + + table.add_row("Environment", f"{active_name} ({env.label})") + table.add_row("API endpoint base", env.publisher_base) + table.add_row("Name", userinfo.get("name", "-")) + table.add_row("Username", userinfo.get("preferred_username", "-")) + table.add_row("Email", userinfo.get("email", "-")) + table.add_row("UserId", userinfo.get("sub", "-")) + + if tokens.scope: + table.add_row("Scopes", tokens.scope) + if tokens.expires_at: + import datetime + exp = datetime.datetime.fromtimestamp(tokens.expires_at).strftime("%Y-%m-%d %H:%M:%S") + table.add_row("Token expires", exp) + + console.print() + console.print(Panel(table, title="[bold]Account[/bold]", border_style="blue", padding=(1, 2))) + console.print() + + +auth_group.add_command(login) +auth_group.add_command(logout) +auth_group.add_command(account_show) \ No newline at end of file diff --git a/ce/commands/config.py b/ce/commands/config.py new file mode 100644 index 0000000..0eb523e --- /dev/null +++ b/ce/commands/config.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import click +from rich.console import Console +from rich.table import Table, box + +from ce.config.context import load_cli_config, save_cli_config +from ce.utils.output import OUTPUT_FORMATS + +console = Console() +err_console = Console(stderr=True) + + +_SETTABLE_KEYS = { + "output": "output_format", +} + +_KEY_HELP = { + "output": f"Default output format. One of: {', '.join(OUTPUT_FORMATS)}", +} + + +@click.group("config") +def config_group() -> None: + """Manage CE CLI configuration.""" + + +@click.command("set") +@click.argument("assignments", nargs=-1, required=True, metavar="KEY=VALUE …") +def config_set(assignments: tuple[str, ...]) -> None: + + cli_config = load_cli_config() + changed: list[str] = [] + errors: list[str] = [] + + for assignment in assignments: + if "=" not in assignment: + errors.append(f"Invalid format '{assignment}' — expected KEY=VALUE.") + continue + + key, _, value = assignment.partition("=") + key = key.strip().lower() + value = value.strip() + + if key not in _SETTABLE_KEYS: + errors.append( + f"Unknown key '{key}'. Valid keys: {', '.join(sorted(_SETTABLE_KEYS))}." + ) + continue + + field = _SETTABLE_KEYS[key] + + # Validate + if key == "output" and value not in OUTPUT_FORMATS: + errors.append( + f"Invalid value '{value}' for 'output'. Must be one of: {', '.join(OUTPUT_FORMATS)}." + ) + continue + + setattr(cli_config, field, value) + changed.append(f"{key} = {value}") + + for err in errors: + err_console.print(f"[red]✗[/red] {err}") + + if changed: + save_cli_config(cli_config) + for line in changed: + console.print(f"[green]✓[/green] {line}") + + if errors: + raise SystemExit(1) + + + +@click.command("get") +@click.argument("key") +def config_get(key: str) -> None: + """Print the value of a configuration key.""" + key = key.strip().lower() + cli_config = load_cli_config() + + if key not in _SETTABLE_KEYS: + err_console.print( + f"[red]✗[/red] Unknown key '{key}'. Valid keys: {', '.join(sorted(_SETTABLE_KEYS))}." + ) + raise SystemExit(1) + + field = _SETTABLE_KEYS[key] + value = getattr(cli_config, field, None) + click.echo(value if value is not None else "") + + +@click.command("list") +def config_list() -> None: + """List all CLI configuration values.""" + cli_config = load_cli_config() + + table = Table(box=box.SIMPLE_HEAD, header_style="bold", border_style="dim") + table.add_column("Key") + table.add_column("Value") + table.add_column("Description", style="dim") + + for key, field in sorted(_SETTABLE_KEYS.items()): + value = getattr(cli_config, field, "") + table.add_row(key, str(value) if value else "[dim]—[/dim]", _KEY_HELP.get(key, "")) + + console.print() + console.print(table) + console.print() + + + +@click.command("reset") +@click.option("--yes", "-y", is_flag=True, help="Skip confirmation.") +def config_reset(yes: bool) -> None: + """Reset all CLI configuration to defaults.""" + if not yes: + click.confirm(" Reset all CLI configuration to defaults?", abort=True) + + from ce.config.context import CLIConfig, save_cli_config + current = load_cli_config() + # Preserve environments and active env; only reset preferences + reset = CLIConfig( + current_environment=current.current_environment, + environments=current.environments, + ) + save_cli_config(reset) + console.print("[green]✓[/green] Configuration reset to defaults.") + + + +config_group.add_command(config_set, name="set") +config_group.add_command(config_get, name="get") +config_group.add_command(config_list, name="list") +config_group.add_command(config_reset, name="reset") diff --git a/ce/commands/env.py b/ce/commands/env.py new file mode 100644 index 0000000..0a3b0ca --- /dev/null +++ b/ce/commands/env.py @@ -0,0 +1,190 @@ +"""Environment command group: ce env list/add/use/show/remove.""" + +from __future__ import annotations + +import click +from rich.console import Console +from rich.panel import Panel +from rich.table import Table, box + +from ce.config.context import ( + DEFAULT_ENVIRONMENTS, + Environment, + get_current_environment, + load_cli_config, + save_cli_config, +) + +console = Console() +err_console = Console(stderr=True) + + +@click.group("env") +def env_group() -> None: + """Manage Credential Engine environments (dev, sandbox, prod, …).""" + + + +@click.command("list") +def env_list() -> None: + """List all configured environments.""" + config = load_cli_config() + + table = Table(box=box.SIMPLE_HEAD, header_style="bold", border_style="dim") + table.add_column("") + table.add_column("Name") + table.add_column("Label") + table.add_column("Publisher base") + + for name, env in sorted(config.environments.items()): + active = "[green]●[/green]" if name == config.current_environment else " " + bold = name == config.current_environment + table.add_row( + active, + f"[bold]{name}[/bold]" if bold else name, + env.label, + env.publisher_base, + ) + + console.print() + console.print(table) + if config.current_environment: + console.print(f" Active: [bold green]{config.current_environment}[/bold green]") + console.print() + + + +@click.command("add") +@click.argument("name") +@click.option("--label", default=None, help="Human-readable label.") +@click.option("--client-id", default=None, metavar="ID", help="OIDC client ID.") +@click.option("--device-auth-url",default=None, metavar="URL", help="Device authorization endpoint.") +@click.option("--token-url", default=None, metavar="URL", help="Token endpoint.") +@click.option("--scopes", default=None, help="Space-separated OIDC scopes.") +@click.option("--publisher-base", default=None, metavar="URL", help="Publisher service base URL.") +@click.option("--no-ssl-verify", is_flag=True, default=False, help="Disable SSL verification.") +@click.option("--activate", "-a", is_flag=True, default=False, help="Set as active immediately.") +def env_add(name, label, client_id, device_auth_url, token_url, scopes, + publisher_base, no_ssl_verify, activate): + config = load_cli_config() + existing = name in config.environments + base = DEFAULT_ENVIRONMENTS.get(name) or config.environments.get(name) + + if base is None and not all([label, client_id, device_auth_url, token_url, publisher_base]): + err_console.print( + f"[red]✗[/red] '{name}' is not a built-in environment. " + "Provide all required options: --label, --client-id, --device-auth-url, " + "--token-url, --publisher-base." + ) + raise SystemExit(1) + + config.environments[name] = Environment( + label= label or (base.label if base else name), + client_id= client_id or (base.client_id if base else ""), + device_auth_url= device_auth_url or (base.device_auth_url if base else ""), + token_url= token_url or (base.token_url if base else ""), + scopes= scopes or (base.scopes if base else "openid profile"), + publisher_base= publisher_base or (base.publisher_base if base else ""), + ssl_verify= not no_ssl_verify, + ) + if activate or config.current_environment is None: + config.current_environment = name + + save_cli_config(config) + verb = "Updated" if existing else "Added" + console.print(f"[green]✓[/green] {verb} environment [bold]{name}[/bold]") + if config.current_environment == name: + console.print(" [dim]Now active.[/dim]") + + +@click.command("use") +@click.argument("name") +def env_use(name: str) -> None: + """Set NAME as the active environment.""" + config = load_cli_config() + if name not in config.environments: + err_console.print( + f"[red]✗[/red] Unknown environment [bold]{name}[/bold]. " + "Run [bold]ce env list[/bold] to see available environments." + ) + raise SystemExit(1) + config.current_environment = name + save_cli_config(config) + env = config.environments[name] + console.print(f"[green]✓[/green] Switched to [bold]{name}[/bold] ({env.label}).") + + +@click.command("show") +@click.argument("name", required=False) +def env_show(name: str | None) -> None: + """Show all details for an environment (defaults to active).""" + config = load_cli_config() + target = name or config.current_environment + + if not target: + err_console.print("[yellow]⚠[/yellow] No active environment.") + return + + env = config.environments.get(target) + if not env: + err_console.print(f"[red]✗[/red] Environment [bold]{target}[/bold] not found.") + raise SystemExit(1) + + table = Table(show_header=False, box=None, padding=(0, 2)) + table.add_column("Key", style="dim") + table.add_column("Value", style="bold") + + active_label = " [green](active)[/green]" if target == config.current_environment else "" + rows = [ + ("Name", f"{target}{active_label}"), + ("Label", env.label), + ("Client ID", env.client_id), + ("Device auth URL", env.device_auth_url), + ("Token URL", env.token_url), + ("Scopes", env.scopes), + ("Publisher base", env.publisher_base), + ("SSL verify", str(env.ssl_verify)), + ] + for k, v in rows: + table.add_row(k, v) + + console.print() + console.print(Panel(table, title=f"[bold]Environment: {target}[/bold]", + border_style="blue", padding=(1, 2))) + console.print() + + + +@click.command("remove") +@click.argument("name") +@click.option("--yes", "-y", is_flag=True, help="Skip confirmation.") +def env_remove(name: str, yes: bool) -> None: + """Remove environment NAME.""" + config = load_cli_config() + if name not in config.environments: + err_console.print(f"[red]✗[/red] Environment [bold]{name}[/bold] not found.") + raise SystemExit(1) + + if not yes: + click.confirm(f" Remove environment '{name}'?", abort=True) + + del config.environments[name] + if config.current_environment == name: + config.current_environment = next(iter(config.environments), None) + if config.current_environment: + console.print(f" [dim]Switched active environment to " + f"[bold]{config.current_environment}[/bold].[/dim]") + + save_cli_config(config) + console.print(f"[green]✓[/green] Removed environment [bold]{name}[/bold].") + + +# --------------------------------------------------------------------------- +# Wire up +# --------------------------------------------------------------------------- + +env_group.add_command(env_list, name="list") +env_group.add_command(env_add, name="add") +env_group.add_command(env_use, name="use") +env_group.add_command(env_show, name="show") +env_group.add_command(env_remove, name="remove") \ No newline at end of file diff --git a/ce/commands/iir.py b/ce/commands/iir.py new file mode 100644 index 0000000..d7672f6 --- /dev/null +++ b/ce/commands/iir.py @@ -0,0 +1,167 @@ +"""IIR command group: ce iir bulk-upload-dids sign-challenges / publish.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import click +from rich.console import Console + +from ce.auth.token_manager import require_login +from ce.config.context import require_environment +from ce.iir.csv_processor import run_generate, run_upload +from ce.iir.did_ops import extract_user_id_from_token +from ce.utils.errors import handle_api_errors + +console = Console() +err_console = Console(stderr=True) + + +@click.group("iir") +def iir_group() -> None: + """IIR (Issuer Identity Registry) commands.""" + + +@iir_group.group("bulk-upload-dids") +def bulk_upload_dids_group() -> None: + """Bulk DID challenge generation and issuer publishing.""" + + +@bulk_upload_dids_group.command("sign-challenges") +@click.option( + "--csv", "csv_path", + required=True, + type=click.Path(exists=True, dir_okay=False, readable=True), + help="Input CSV file (columns: CTID, DID, VerificationMethod, Algorithm, PrivateKey, ValidFrom, ValidUntil).", +) +@click.option( + "--output", "-o", "output_path", + default=None, + help="Output CSV path. Default: Output--.csv", +) +@click.option( + "--errors", "errors_path", + default=None, + help="Errors CSV path. Default: Errors--.csv", +) +@handle_api_errors +def generate(csv_path: str, output_path: str | None, errors_path: str | None) -> None: + """Check membership & registry, validate DIDs, create challenges, sign JWTs. + + \b + Checks per row: + 1. CTID format + 2. Date format validation (ValidFrom, ValidUntil if provided) + 3. Membership check - Is User a member of the org + 4. Registry check - Is CTID published to the registry + 5. DID classification + 6. DID validation via API + 7. Verification method + 8. Create challenge + 9. Sign JWT + 10. Local JWT verification + + \b + Input CSV columns (required): + CTID, DID, VerificationMethod, Algorithm, PrivateKey + + \b + Input CSV columns (optional): + ValidFrom, ValidUntil (format: MM/DD/YYYY) + + \b + Produces two files: + Output--.csv — signed rows ready for publish + Errors--.csv — rows that failed, fix and rerun + + \b + Examples: + ce iir bulk-upload-dids sign-challenges --csv input.csv + ce iir bulk-upload-dids sign-challenges --csv input.csv --output signed.csv + """ + access_token = require_login() + _, env = require_environment() + + try: + user_id = extract_user_id_from_token(access_token) + except ValueError as e: + err_console.print(f"[red]✗[/red] {e}") + raise SystemExit(1) + + stem = Path(csv_path).stem + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + resolved_output = output_path or f"Output-{stem}-{timestamp}.csv" + resolved_errors = errors_path or f"Errors-{stem}-{timestamp}.csv" + + console.print(f"\n[dim]Environment:[/dim] [bold]{env.label}[/bold] ({env.publisher_base})") + console.print(f"[dim]User ID:[/dim] [bold]{user_id}[/bold]") + + ok, errors = run_generate( + input_path=csv_path, + output_path=resolved_output, + errors_path=resolved_errors, + access_token=access_token, + env=env, + user_id=user_id, + ) + + if ok > 0: + console.print( + f"\n[dim]Next step:[/dim] " + f"[bold]ce iir bulk-upload-dids publish --output-file \"{resolved_output}\"[/bold]" + ) + + + +@bulk_upload_dids_group.command("publish") +@click.option( + "--output-file", "output_path", + required=True, + type=click.Path(exists=True, dir_okay=False, readable=True), + help="Output CSV produced by the sign-challenges step.", +) +@click.option( + "--publish-errors", "publish_errors_path", + default=None, + help="Publish errors CSV path. Default: PublishErrors--.csv", +) +@click.option( + "--yes", "-y", + is_flag=True, + default=False, + help="Skip confirmation prompt.", +) +@handle_api_errors +def upload(output_path: str, publish_errors_path: str | None, yes: bool) -> None: + """Verify JWT signatures and publish issuers. + + \b + Steps per row: + 1. JWT signature verify + 2. Save Token + 3. Submit to IIR + + + \b + Examples: + ce iir bulk-upload-dids publish --output-file Output-input-20240101T000000Z.csv + ce iir bulk-upload-dids publish --output-file signed.csv --yes + """ + access_token = require_login() + _, env = require_environment() + + stem = Path(output_path).stem + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + resolved_pub_errors = publish_errors_path or f"PublishErrors-{stem}-{timestamp}.csv" + + console.print(f"\n[dim]Environment:[/dim] [bold]{env.label}[/bold] ({env.publisher_base})") + + run_upload( + output_path=output_path, + publish_errors_path=resolved_pub_errors, + access_token=access_token, + env=env, + yes=yes, + ) \ No newline at end of file diff --git a/ce/commands/version.py b/ce/commands/version.py new file mode 100644 index 0000000..897fb7a --- /dev/null +++ b/ce/commands/version.py @@ -0,0 +1,59 @@ +"""Version command: ce version. + +Prints version info for the CLI and (optionally) checks PyPI for a newer release. +""" + +from __future__ import annotations + +import click +from rich.console import Console + +from ce import __version__ + +console = Console() +err_console = Console(stderr=True) + + +@click.command("version") +@click.option( + "--check", + is_flag=True, + default=False, + help="Check for a newer version on PyPI.", +) +def version_command(check: bool) -> None: + """Show the current CLI version.""" + console.print(f"ce version [bold]{__version__}[/bold]") + + if check: + _check_for_update() + + +def _check_for_update() -> None: + """Query PyPI for the latest published version of ce-cli.""" + import httpx + + try: + with console.status("[dim]Checking for updates…[/dim]"): + resp = httpx.get( + "https://pypi.org/pypi/ce-cli/json", + timeout=5, + follow_redirects=True, + ) + if not resp.is_success: + console.print("[dim]Could not reach PyPI.[/dim]") + return + + latest = resp.json()["info"]["version"] + except Exception as exc: + console.print(f"[dim]Update check failed: {exc}[/dim]") + return + + if latest == __version__: + console.print(f"[green]✓[/green] You are up to date.") + else: + console.print( + f"[yellow]⚠[/yellow] A newer version is available: " + f"[bold]{latest}[/bold] (you have {__version__})\n" + f" Upgrade with: [bold]pip install --upgrade ce-cli[/bold]" + ) \ No newline at end of file diff --git a/ce/config/__init__.py b/ce/config/__init__.py new file mode 100644 index 0000000..81413a5 --- /dev/null +++ b/ce/config/__init__.py @@ -0,0 +1,29 @@ +from ce.config.settings import ( + StoredTokens, + delete_tokens, + ensure_config_dir, + load_tokens, + save_tokens, +) +from ce.config.context import ( + CLIConfig, + Environment, + get_current_environment, + load_cli_config, + require_environment, + save_cli_config, +) + +__all__ = [ + "StoredTokens", + "delete_tokens", + "ensure_config_dir", + "load_tokens", + "save_tokens", + "CLIConfig", + "Environment", + "get_current_environment", + "load_cli_config", + "require_environment", + "save_cli_config", +] \ No newline at end of file diff --git a/ce/config/context.py b/ce/config/context.py new file mode 100644 index 0000000..eb61189 --- /dev/null +++ b/ce/config/context.py @@ -0,0 +1,178 @@ +"""Named environment (context) management.""" + +from __future__ import annotations + +import json +from typing import Optional + +from pydantic import BaseModel, Field + +from ce.config.settings import CONFIG_FILE, ensure_config_dir + + +class Environment(BaseModel): + """A named Credential Engine deployment target.""" + + label: str = Field(description="environment.") + client_id: str = Field(description="Client ID.") + device_auth_url: str = Field(description="Keycloak device authorization endpoint.") + token_url: str = Field(description="Keycloak token endpoint.") + scopes: str = Field(default="openid profile", description="Space-separated OIDC scopes.") + publisher_base: str = Field(description="Publisher base URL (BFF for all API calls).") + ssl_verify: bool = Field(default=True, description="Verify SSL certificates.") + + +class CLIConfig(BaseModel): + """Top-level CLI config file model.""" + + current_environment: Optional[str] = None + environments: dict[str, Environment] = Field(default_factory=dict) + + +DEFAULT_ENVIRONMENTS: dict[str, Environment] = { + "dev": Environment( + label="Development", + client_id="IIR-TokenGenerator-CLI", + device_auth_url="http://localhost:8080/realms/CE_Accounts/protocol/openid-connect/auth/device", + token_url="http://localhost:8080/realms/CE_Accounts/protocol/openid-connect/token", + scopes="openid profile", + publisher_base="https://localhost:44330", + ssl_verify=False, + ), + "sandbox": Environment( + label="Sandbox", + client_id="IIR-TokenGenerator-CLI", + device_auth_url="https://login.sandbox.credentialengine.org/realms/CE-Sandbox/protocol/openid-connect/auth/device", + token_url="https://login.sandbox.credentialengine.org/realms/CE-Sandbox/protocol/openid-connect/token", + scopes="openid profile", + publisher_base="https://sandbox.credentialengine.org/publisher", + ssl_verify=True, + ), + "prod": Environment( + label="Production", + client_id="IIR-TokenGenerator-CLI", + device_auth_url="https://login.credentialengine.org/realms/CE-Prod/protocol/openid-connect/auth/device", + token_url="https://login.credentialengine.org/realms/CE-Prod/protocol/openid-connect/token", + scopes="openid profile", + publisher_base="https://apps.credentialengine.org/publisher", + ssl_verify=True, + ), +} + + +def _seed_environments(config: CLIConfig) -> tuple[CLIConfig, bool]: + """Populate any missing default environments (non-destructive, never sets active).""" + changed = False + for name, env in DEFAULT_ENVIRONMENTS.items(): + if name not in config.environments: + config.environments[name] = env + changed = True + return config, changed + + +def prompt_select_environment() -> str: + """Interactively ask the user to pick an environment. Returns the chosen name.""" + from rich.console import Console + from rich.panel import Panel + from rich.table import Table, box + + console = Console() + + console.print() + console.print(Panel( + "[bold]Welcome to the Credential Engine CLI![/bold]\n\n" + "Please select the environment you want to connect to.\n" + "[dim]You can change this at any time with [bold]ce env use [/bold][/dim]", + border_style="blue", + padding=(1, 2), + )) + + table = Table(box=box.SIMPLE_HEAD, header_style="bold", border_style="dim", padding=(0, 2)) + table.add_column("#", style="dim", width=3) + table.add_column("Name") + table.add_column("Label") + table.add_column("Publisher base") + + names = list(DEFAULT_ENVIRONMENTS.keys()) + for i, (name, env) in enumerate(DEFAULT_ENVIRONMENTS.items(), 1): + table.add_row(str(i), f"[bold]{name}[/bold]", env.label, env.publisher_base) + + console.print(table) + + while True: + raw = console.input( + " [bold]Select environment[/bold] [dim](enter name or number, e.g. sandbox)[/dim]: " + ).strip().lower() + + if raw.isdigit(): + idx = int(raw) - 1 + if 0 <= idx < len(names): + chosen = names[idx] + break + console.print(f" [red]✗[/red] Please enter a number between 1 and {len(names)}.") + continue + + if raw in DEFAULT_ENVIRONMENTS: + chosen = raw + break + + console.print( + f" [red]✗[/red] '{raw}' is not a valid environment. " + f"Choose from: {', '.join(names)}." + ) + + env = DEFAULT_ENVIRONMENTS[chosen] + console.print(f"\n [green]✓[/green] Environment set to [bold]{chosen}[/bold] ({env.label}).\n") + return chosen + + + +def load_cli_config() -> CLIConfig: + + first_run = not CONFIG_FILE.exists() + + if first_run: + config = CLIConfig() + config, _ = _seed_environments(config) + chosen = prompt_select_environment() + config.current_environment = chosen + save_cli_config(config) + return config + + try: + data = json.loads(CONFIG_FILE.read_text()) + config = CLIConfig(**data) + except Exception: + config = CLIConfig() + + config, changed = _seed_environments(config) + if changed: + save_cli_config(config) + return config + + +def save_cli_config(config: CLIConfig) -> None: + """Persist CLI config to disk.""" + ensure_config_dir() + CONFIG_FILE.write_text(config.model_dump_json(indent=2)) + CONFIG_FILE.chmod(0o600) + + +def get_current_environment() -> Optional[tuple[str, Environment]]: + """Return (name, Environment) for the active environment, or None.""" + config = load_cli_config() + name = config.current_environment + if name and name in config.environments: + return name, config.environments[name] + return None + + +def require_environment() -> tuple[str, Environment]: + """Return the active environment or raise a UsageError.""" + import click + result = get_current_environment() + if result is None: + raise click.UsageError( + "No environment selected. Run 'ce env use ' to activate one." + ) + return result \ No newline at end of file diff --git a/ce/config/settings.py b/ce/config/settings.py new file mode 100644 index 0000000..77b18dd --- /dev/null +++ b/ce/config/settings.py @@ -0,0 +1,119 @@ +"""Configuration management for the Credential Engine CLI.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Optional + +from pydantic import BaseModel, Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +# --------------------------------------------------------------------------- +# Paths +# --------------------------------------------------------------------------- + +def _config_dir() -> Path: + """Return the CE config directory, honouring $CE_CONFIG_DIR override.""" + override = os.environ.get("CE_CONFIG_DIR") + if override: + return Path(override) + return Path.home() / ".ce" + + +CONFIG_DIR = _config_dir() +CONFIG_FILE = CONFIG_DIR / "config.json" +TOKEN_FILE = CONFIG_DIR / "tokens.json" + + +class OIDCSettings(BaseSettings): + """Keycloak / OIDC connection settings. + + These can be overridden by environment variables prefixed with CE_. + + Example .env or shell exports:: + + CE_OIDC_ISSUER=https://auth.example.com/realms/myrealm + CE_OIDC_CLIENT_ID=ce-cli + """ + + model_config = SettingsConfigDict( + env_prefix="CE_OIDC_", + env_file=str(CONFIG_DIR / ".env"), + env_file_encoding="utf-8", + extra="ignore", + ) + + issuer: str = Field( + default="https://auth.example.com/realms/credential-engine", + description="Keycloak realm issuer URL.", + ) + client_id: str = Field( + default="ce-cli", + description="OIDC public client ID registered in Keycloak.", + ) + scopes: str = Field(default="openid profile email offline_access") + + + @property + def device_authorization_endpoint(self) -> str: + return f"{self.issuer.rstrip('/')}/protocol/openid-connect/auth/device" + + @property + def token_endpoint(self) -> str: + return f"{self.issuer.rstrip('/')}/protocol/openid-connect/token" + + @property + def userinfo_endpoint(self) -> str: + return f"{self.issuer.rstrip('/')}/protocol/openid-connect/userinfo" + + @property + def end_session_endpoint(self) -> str: + return f"{self.issuer.rstrip('/')}/protocol/openid-connect/logout" + + +class StoredTokens(BaseModel): + """OIDC tokens persisted to disk.""" + + access_token: str + refresh_token: Optional[str] = None + id_token: Optional[str] = None + token_type: str = "Bearer" + expires_at: Optional[float] = None # Unix timestamp + scope: Optional[str] = None + + +def ensure_config_dir() -> None: + """Create the CE config directory with safe permissions.""" + CONFIG_DIR.mkdir(mode=0o700, parents=True, exist_ok=True) + + +def load_tokens() -> Optional[StoredTokens]: + """Load stored tokens from disk, or return None if not present.""" + if not TOKEN_FILE.exists(): + return None + try: + data = json.loads(TOKEN_FILE.read_text()) + return StoredTokens(**data) + except Exception: + return None + + +def save_tokens(tokens: StoredTokens) -> None: + """Persist tokens to disk with restrictive permissions.""" + ensure_config_dir() + TOKEN_FILE.write_text(tokens.model_dump_json(indent=2)) + TOKEN_FILE.chmod(0o600) + + +def delete_tokens() -> None: + """Remove stored tokens from disk.""" + if TOKEN_FILE.exists(): + TOKEN_FILE.unlink() + + +def load_oidc_settings() -> OIDCSettings: + """Return OIDC settings, layering env vars over defaults.""" + return OIDCSettings() diff --git a/ce/iir/__init__.py b/ce/iir/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ce/iir/csv_processor.py b/ce/iir/csv_processor.py new file mode 100644 index 0000000..2c5266a --- /dev/null +++ b/ce/iir/csv_processor.py @@ -0,0 +1,367 @@ +"""CSV processing for bulk DID upload. + sign-challenges + CTID Validation + registry lookup, + DID validation, + challenge creation, + JWT signing + Output CSV + publish + JWT signature verify + Save Token + publish issuer +""" + +from __future__ import annotations + +import csv +from pathlib import Path + +from rich.console import Console +from rich.table import Table, box + +from ce.iir.did_ops import ( + call_is_user_member, + call_create_challenge, + call_get_registry_resource, + call_save_challenge_token, + call_validate_did_key, + call_validate_did_web, + call_verify_jwt_signature, + classify_did, + sign_proof_jwt, + validate_ctid, + validate_date, + validate_did_key_prefix, + validate_verification_method, + verify_proof_jwt, + call_submit_to_iir +) + +console = Console() +err_console = Console(stderr=True) + + +OUTPUT_FIELDS = ["CTID", "DID", "VerificationMethod", "Challenge", "ProofJWT", "ValidFrom", "ValidUntil"] +ERROR_FIELDS = ["CTID", "DID", "VerificationMethod", "Algorithm", "Error"] +PUBLISH_ERR_FIELDS = ["CTID", "DID", "VerificationMethod", "Challenge", "Error"] + + +def run_generate( + input_path: str, + output_path: str, + errors_path: str, + access_token: str, + env, + user_id: str, +) -> tuple[int, int]: + """Validate membership, registry, DIDs, create challenges, sign JWTs. + + Steps per row: + 1. Validate CTID format + 2. Membership check — is the user a member of this org? + 3. Registry check — is the CTID published in the registry? + 4. Classify DID + prefix check + 5. DID API validation + 6. Validate verification method + 7. Create challenge + 8. Sign JWT + 9. Local JWT verification + + Returns (ok_count, error_count). + """ + input_file = Path(input_path) + output_file = Path(output_path) + errors_file = Path(errors_path) + + if not input_file.exists(): + raise FileNotFoundError(f"Input file not found: {input_path}") + + with input_file.open(newline="", encoding="utf-8-sig") as f: + reader = csv.DictReader(f) + norm_map = {field: field.strip().lower() for field in (reader.fieldnames or [])} + rows = [{norm_map[k]: v.strip() for k, v in row.items()} for row in reader] + + missing = [c for c in ("ctid", "did", "verificationmethod") if c not in norm_map.values()] + if missing: + raise ValueError(f"Missing required CSV column(s): {', '.join(missing)}") + + ok_rows : list[dict] = [] + error_rows : list[dict] = [] + total = len(rows) + + table = Table(box=box.SIMPLE_HEAD, header_style="bold", border_style="dim", show_lines=False) + table.add_column("#", width=4, style="dim") + table.add_column("CTID", width=44) + table.add_column("Member", width=8) + table.add_column("Registry", width=10) + table.add_column("DID type", width=10) + table.add_column("Result", width=36) + + console.print(f"\n[bold]Processing {total} row(s) from '{input_path}'…[/bold]\n") + + for i, row in enumerate(rows, 1): + ctid = row.get("ctid", "") + did = row.get("did", "") + vm = row.get("verificationmethod", "") + alg = row.get("algorithm", "") + privkey = row.get("privatekey", "") + valid_from = row.get("validfrom", "") + valid_until = row.get("validuntil", "") + did_type = "unknown" + error = "" + member_icon = "[dim]—[/dim]" + registry_icon = "[dim]—[/dim]" + registry_data : dict = {} + + def fail(reason: str) -> None: + nonlocal error + error = reason + + ctid_ok, ctid_err = validate_ctid(ctid) + if not ctid_ok: + fail(ctid_err) + + if not error: + ok, err, valid_from = validate_date(valid_from, "ValidFrom") + if not ok: + fail(err) + if not error: + ok, err, valid_until = validate_date(valid_until, "ValidUntil") + if not ok: + fail(err) + + if not error: + ok, err = call_is_user_member( + user_id, ctid, access_token, env.publisher_base, env.ssl_verify + ) + member_icon = "[green]✓[/green]" if ok else "[red]✗[/red]" + if not ok: + fail(err) + + if not error: + ok, registry_data, err = call_get_registry_resource( + ctid, access_token, env.publisher_base, env.ssl_verify + ) + registry_icon = "[green]✓[/green]" if ok else "[red]✗[/red]" + if not ok: + fail(err) + + if not error: + did_type, type_err = classify_did(did) + if type_err: + fail(type_err) + elif did_type == "did:key": + ok, err = validate_did_key_prefix(did, alg) + if not ok: + fail(err) + + api_vm_ids: list[str] = [] + if not error: + if did_type == "did:key": + ok, data, err = call_validate_did_key(did, alg, access_token, env.publisher_base, env.ssl_verify) + else: + ok, data, err = call_validate_did_web(did, access_token, env.publisher_base, env.ssl_verify) + if not ok: + fail(err) + else: + api_vm_ids = data.get("verificationMethodIds") or [] + + if not error: + ok, err = validate_verification_method(vm, did, api_vm_ids) + if not ok: + fail(err) + + challenge_uuid = "" + challenge_payload = {} + if not error: + ok, challenge, err = call_create_challenge( + ctid, vm, access_token, env.publisher_base, env.ssl_verify + ) + if not ok: + fail(err) + else: + challenge_uuid = challenge.get("challenge") or challenge.get("Challenge", "") + challenge_payload = { + "did": challenge.get("Did") or challenge.get("did", did), + "challenge": challenge_uuid, + "aud": challenge.get("Aud") or challenge.get("aud", ""), + "iat": challenge.get("Iat") or challenge.get("iat", 0), + "exp": challenge.get("Exp") or challenge.get("exp", 0), + "ctid": ctid, + } + + proof_jwt = "" + if not error: + if not privkey: + fail("PrivateKey is required for JWT signing") + else: + try: + proof_jwt = sign_proof_jwt(privkey, vm, challenge_payload) + except Exception as e: + fail(f"JWT signing failed: {e}") + + if not error: + ok, err = verify_proof_jwt(proof_jwt, vm) + if not ok: + fail(err) + + ctid_short = (ctid[:42] + "…") if len(ctid) > 43 else ctid + + if not error: + result = f"[green]✓[/green] {challenge_uuid[:20]}…" + ok_rows.append({ + "CTID": ctid, + "DID": did, + "VerificationMethod": vm, + "Challenge": challenge_uuid, + "ProofJWT": proof_jwt, + "ValidFrom": valid_from, + "ValidUntil": valid_until, + }) + else: + result = f"[red]✗[/red] {error[:36]}" + error_rows.append({ + "CTID": ctid, + "DID": did, + "VerificationMethod": vm, + "Algorithm": alg, + "Error": error, + }) + + table.add_row(str(i), ctid_short, member_icon, registry_icon, did_type, result) + + console.print(table) + + with output_file.open("w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=OUTPUT_FIELDS) + writer.writeheader() + writer.writerows(ok_rows) + + with errors_file.open("w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=ERROR_FIELDS) + writer.writeheader() + writer.writerows(error_rows) + + console.print(f"\n[green]✓[/green] {len(ok_rows)} succeeded → [bold]{output_file}[/bold]") + if error_rows: + console.print(f"[red]✗[/red] {len(error_rows)} failed → [bold]{errors_file}[/bold]") + console.print( + f" Fix errors and rerun: " + f"[bold]ce iir bulk-upload-dids sign-challenges --csv {errors_file.name}[/bold]" + ) + + return len(ok_rows), len(error_rows) + + +def run_upload( + output_path: str, + publish_errors_path: str, + access_token: str, + env, + yes: bool = False, +) -> None: + + output_file = Path(output_path) + publish_err_file = Path(publish_errors_path) + + if not output_file.exists() or output_file.stat().st_size == 0: + console.print("\nOutput file is empty or missing — nothing to upload.") + return + + with output_file.open(newline="", encoding="utf-8-sig") as f: + rows = list(csv.DictReader(f)) + + if not rows: + console.print("Output CSV has no rows.") + return + + if not yes: + import click + click.confirm(f"\n Publish {len(rows)} issuer(s) from '{output_path}'?", abort=True) + + publish_error_rows: list[dict] = [] + + table = Table(box=box.SIMPLE_HEAD, header_style="bold", border_style="dim") + table.add_column("#", width=4, style="dim") + table.add_column("CTID", width=44) + table.add_column("JWT", width=6) + table.add_column("Publish", width=9) + + console.print(f"\n[bold]Publishing {len(rows)} issuer(s)…[/bold]\n") + + for i, row in enumerate(rows, 1): + ctid = row["CTID"] + did = row["DID"] + proof_jwt = row["ProofJWT"] + challenge_uuid = row["Challenge"] + vm = row["VerificationMethod"] + valid_from = row.get("ValidFrom", "") + valid_until = row.get("ValidUntil", "") + ctid_short = (ctid[:42] + "…") if len(ctid) > 43 else ctid + pub_error = "" + jwt_icon = "[dim]—[/dim]" + publish_icon = "[dim]—[/dim]" + + def pub_fail(reason: str) -> None: + nonlocal pub_error + pub_error = reason + + ok, err = call_verify_jwt_signature( + proof_jwt, challenge_uuid, access_token, env.publisher_base, env.ssl_verify + ) + jwt_icon = "[green]✓[/green]" if ok else "[red]✗[/red]" + if not ok: + pub_fail(err) + + if not pub_error: + ok, err = call_save_challenge_token( + ctid, challenge_uuid, proof_jwt, + access_token, env.publisher_base, env.ssl_verify, + ) + if not ok: + pub_fail(err) + + if not pub_error: + ok, registry_data, err = call_get_registry_resource( + ctid, access_token, env.publisher_base, env.ssl_verify + ) + if not ok: + pub_fail(err) + else: + ok, err = call_submit_to_iir( + ctid, did, + registry_data.get("Name", ""), + registry_data.get("LegalName", ""), + registry_data.get("CredentialRegistryUri", ""), + registry_data.get("SubjectWebpage", ""), + access_token, env.publisher_base, env.ssl_verify, + valid_from=valid_from, + valid_until=valid_until, + ) + publish_icon = "[green]✓[/green]" if ok else "[red]✗[/red]" + if not ok: + pub_fail(err) + + if pub_error: + publish_error_rows.append({ + "CTID": ctid, "DID": did, + "VerificationMethod": vm, "Challenge": challenge_uuid, + "Error": pub_error, + }) + + table.add_row(str(i), ctid_short, jwt_icon, publish_icon) + + console.print(table) + + with publish_err_file.open("w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=PUBLISH_ERR_FIELDS) + writer.writeheader() + writer.writerows(publish_error_rows) + + ok_count = len(rows) - len(publish_error_rows) + console.print(f"\n[green]✓[/green] {ok_count} published successfully.") + if publish_error_rows: + console.print( + f"[red]✗[/red] {len(publish_error_rows)} error(s) → [bold]{publish_err_file}[/bold]" + ) \ No newline at end of file diff --git a/ce/iir/did_ops.py b/ce/iir/did_ops.py new file mode 100644 index 0000000..1e58788 --- /dev/null +++ b/ce/iir/did_ops.py @@ -0,0 +1,477 @@ +"""API calls and validations""" + +from __future__ import annotations + +import base64 +import json +import re +from typing import Any + +import base58 +import jwt +import requests +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + + +CE_GUID_RE = re.compile( + r"^ce-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", + re.IGNORECASE, +) + +SUPPORTED_ALGORITHMS = {"Ed25519", "secp256k1", "P-256", "X25519"} + +DID_KEY_PREFIX_MAP = { + "z6Mk": "Ed25519", + "zQ3s": "secp256k1", + "zDn": "P-256", + "z6LS": "X25519", +} + +def validate_ctid(ctid: str) -> tuple[bool, str]: + if not ctid: + return False, "CTID is empty" + if CE_GUID_RE.match(ctid): + return True, "" + return False, f"CTID '{ctid}' does not match ce-guid format" + + +def validate_date(value: str, field_name: str) -> tuple[bool, str, str]: + if not value: + return True, "", "" + from datetime import datetime, timezone + try: + parsed = datetime.strptime(value, "%m/%d/%Y") + utc_dt = parsed.replace(tzinfo=timezone.utc) + iso_value = utc_dt.strftime("%Y-%m-%dT%H:%M:%SZ") + return True, "", iso_value + except ValueError: + return False, f"{field_name} must be in MM/DD/YYYY format, got '{value}'", "" + + +def classify_did(did: str) -> tuple[str, str]: + if not did: + return "unknown", "DID is empty" + if did.startswith("did:key:"): + return "did:key", "" + if did.startswith("did:web:"): + return "did:web", "" + return "unknown", f"unrecognised DID method in '{did}'" + + +def validate_did_key_prefix(did: str, algorithm: str) -> tuple[bool, str]: + if not algorithm: + return False, "Algorithm is required for did:key" + if algorithm not in SUPPORTED_ALGORITHMS: + return False, f"unsupported algorithm '{algorithm}' — must be one of {', '.join(sorted(SUPPORTED_ALGORITHMS))}" + key_part = did[len("did:key:"):] + if not key_part: + return False, "did:key has no key material after 'did:key:'" + expected = next( + (alg for pfx, alg in DID_KEY_PREFIX_MAP.items() if key_part.startswith(pfx)), None + ) + if expected is None: + return False, f"unrecognised did:key prefix '{key_part[:8]}...'" + if expected != algorithm: + return False, f"algorithm mismatch: key prefix suggests '{expected}' but CSV declares '{algorithm}'" + return True, "" + + +def validate_verification_method(vm: str, did: str, api_vm_ids: list[str]) -> tuple[bool, str]: + if not vm: + return False, "VerificationMethod is empty" + if not vm.startswith(did.split("#")[0]): + return False, f"VerificationMethod '{vm}' does not belong to DID '{did}'" + if api_vm_ids and vm not in api_vm_ids: + return False, ( + f"VerificationMethod '{vm}' not found in DID document. " + f"Valid: {', '.join(api_vm_ids)}" + ) + return True, "" + + +def _headers(token: str) -> dict: + return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + + +def _error_detail(r: requests.Response) -> str: + try: + body = r.json() + return ( + body.get("message") + or body.get("detail") + or body.get("title") + or body.get("Error") + or r.text + ) + except Exception: + return r.text + + + +def call_is_user_member( + user_id: str, ctid: str, token: str, publisher_base: str, ssl_verify: bool +) -> tuple[bool, str]: + """GET /api/iir/isUserAMember?userId=...&ctid=... + Returns { valid: bool }. + """ + try: + r = requests.get( + f"{publisher_base.rstrip('/')}/api/iir/isUserAMember", + params={"userId": user_id, "ctid": ctid}, + headers=_headers(token), + timeout=60, + verify=ssl_verify, + ) + if not r.ok: + return False, f"isUserMember error {r.status_code}: {_error_detail(r)}" + data = r.json() + if not data.get("valid", False): + return False, "User is not a member of this organization" + return True, "" + except requests.RequestException as e: + return False, f"isUserMember request failed: {e}" + + +def call_get_registry_resource( + ctid: str, token: str, publisher_base: str, ssl_verify: bool +) -> tuple[bool, dict, str]: + try: + r = requests.get( + f"{publisher_base.rstrip('/')}/api/iir/getRegistryResource", + params={"ctid": ctid}, + headers=_headers(token), + timeout=60, + verify=ssl_verify, + ) + if r.status_code == 400: + return False, {}, f"Registry lookup failed: {_error_detail(r)}" + if not r.ok: + return False, {}, f"Registry API error {r.status_code}: {_error_detail(r)}" + data = r.json() + if not data.get("ExistsInRegistry", False): + return False, {}, f"CTID '{ctid}' not found in registry" + return True, data, "" + except requests.RequestException as e: + return False, {}, f"Registry API request failed: {e}" + + +def call_validate_did_key( + did: str, alg: str, token: str, publisher_base: str, ssl_verify: bool +) -> tuple[bool, dict, str]: + try: + r = requests.get( + f"{publisher_base.rstrip('/')}/api/iir/validateDidKey", + params={"did": did, "alg": alg}, + headers=_headers(token), + timeout=60, + verify=ssl_verify, + ) + if r.status_code == 400: + return False, {}, f"DID validation failed: {_error_detail(r)}" + if not r.ok: + return False, {}, f"validateDidKey error {r.status_code}: {_error_detail(r)}" + return True, r.json(), "" + except requests.RequestException as e: + return False, {}, f"validateDidKey request failed: {e}" + + +def call_validate_did_web( + did: str, token: str, publisher_base: str, ssl_verify: bool +) -> tuple[bool, dict, str]: + try: + r = requests.get( + f"{publisher_base.rstrip('/')}/api/iir/validateDidWeb", + params={"did": did}, + headers=_headers(token), + timeout=60, + verify=ssl_verify, + ) + if r.status_code == 400: + return False, {}, f"DID validation failed: {_error_detail(r)}" + if r.status_code == 502: + return False, {}, "could not fetch remote DID document (502)" + if not r.ok: + return False, {}, f"validateDidWeb error {r.status_code}: {_error_detail(r)}" + return True, r.json(), "" + except requests.RequestException as e: + return False, {}, f"validateDidWeb request failed: {e}" + + +def call_create_challenge( + ctid: str, vm_id: str, token: str, publisher_base: str, ssl_verify: bool +) -> tuple[bool, dict, str]: + try: + r = requests.post( + f"{publisher_base.rstrip('/')}/api/iir/createChallenges", + headers=_headers(token), + json={"Ctid": ctid, "VerificationMethodIds": [vm_id]}, + timeout=60, + verify=ssl_verify, + ) + if r.status_code == 400: + return False, {}, f"createChallenges failed: {_error_detail(r)}" + if not r.ok: + return False, {}, f"createChallenges error {r.status_code}: {_error_detail(r)}" + challenges = r.json() + if not challenges: + return False, {}, "createChallenges returned empty array" + return True, challenges[0], "" + except requests.RequestException as e: + return False, {}, f"createChallenges request failed: {e}" + + + +def call_verify_jwt_signature( + proof_jwt: str, challenge_uuid: str, token: str, publisher_base: str, ssl_verify: bool +) -> tuple[bool, str]: + try: + r = requests.post( + f"{publisher_base.rstrip('/')}/api/iir/validateJwt", + headers=_headers(token), + json={"Jwt": proof_jwt, "ChallengeId": challenge_uuid}, + timeout=60, + verify=ssl_verify, + ) + if r.status_code == 400: + return False, f"JWT validation failed: {_error_detail(r)}" + if r.status_code == 404: + return False, "Challenge not found (404)" + if not r.ok: + return False, f"validateJwt error {r.status_code}: {_error_detail(r)}" + return True, "" + except requests.RequestException as e: + return False, f"validateJwt request failed: {e}" + + + + +def call_save_challenge_token( + ctid: str, challenge_uuid: str, token_jwt: str, + token: str, publisher_base: str, ssl_verify: bool +) -> tuple[bool, str]: + try: + r = requests.post( + f"{publisher_base.rstrip('/')}/api/iir/saveChallengeToken", + headers=_headers(token), + json={"Ctid": ctid, "ChallengeId": challenge_uuid, "Token": token_jwt}, + timeout=60, + verify=ssl_verify, + ) + if r.status_code == 400: + return False, f"saveChallengeToken failed: {_error_detail(r)}" + if not r.ok: + return False, f"saveChallengeToken error {r.status_code}: {_error_detail(r)}" + return True, "" + except requests.RequestException as e: + return False, f"saveChallengeToken request failed: {e}" + + + + +def call_submit_to_iir( + ctid: str, + did: str, + name: str, + legal_name: str, + registry_uri: str, + subject_web_page: str, + token: str, + publisher_base: str, + ssl_verify: bool, + logo_uri: str = "", + logo_base64: str = "", + valid_from: str = "", + valid_until: str = "", +) -> tuple[bool, str]: + try: + payload = { + "CTID": ctid, + "DID": did, + "Name": name, + "LegalName": legal_name, + "CredentialRegistryUri": registry_uri, + "SubjectWebPage": subject_web_page, + "LogoBase64": logo_base64 or None, + "LogoUri": logo_uri or None, + } + if valid_from: + payload["ValidFrom"] = valid_from + if valid_until: + payload["ValidUntil"] = valid_until + + r = requests.post( + f"{publisher_base.rstrip('/')}/api/iir/submitToIIR", + headers=_headers(token), + json=payload, + timeout=60, + verify=ssl_verify, + ) + if r.status_code == 409: + return False, "Did already exists (409)" + if r.status_code == 400: + return False, f"submitToIIR failed: {_error_detail(r)}" + if not r.ok: + return False, f"submitToIIR error {r.status_code}: {_error_detail(r)}" + return True, "" + except requests.RequestException as e: + return False, f"submitToIIR request failed: {e}" + + + + +def _parse_uvarint(data: bytes) -> tuple[int, int]: + x, s = 0, 0 + for i, b in enumerate(data): + x |= (b & 0x7F) << s + if (b & 0x80) == 0: + return x, i + 1 + s += 7 + raise ValueError("Invalid varint") + + +def _decode_multibase_base58btc(z_value: str) -> bytes: + if not z_value.startswith("z"): + raise ValueError("Expected multibase base58btc string starting with 'z'") + return base58.b58decode(z_value[1:]) + + +def _extract_ed25519_seed(private_key_multibase: str) -> bytes: + raw = _decode_multibase_base58btc(private_key_multibase.strip()) + if len(raw) == 32: + return raw + if len(raw) == 33 and raw[0] == 0x00: + return raw[1:] + if len(raw) == 34 and raw[0] == 0x00 and raw[1] == 0x20: + return raw[2:] + code, n = _parse_uvarint(raw) + if code == 0x1300 and len(raw) - n == 32: + return raw[n:] + raise ValueError("Unrecognized privateKey encoding") + + +def _extract_ed25519_pub_from_did_key(did_or_kid: str) -> bytes: + base = did_or_kid.split("#", 1)[0].strip() + if not base.startswith("did:key:"): + raise ValueError("Not a did:key DID") + mb = base[len("did:key:"):] + raw = _decode_multibase_base58btc(mb) + if len(raw) == 32: + return raw + code, n = _parse_uvarint(raw) + if code == 0xED and (len(raw) - n) == 32: + return raw[n:] + raise ValueError("Invalid did:key encoding") + + +def _did_web_to_url(did_web: str) -> str: + method_specific = did_web[len("did:web:"):] + parts = method_specific.split(":") + host = parts[0] + path_parts = parts[1:] + if not path_parts: + return f"https://{host}/.well-known/did.json" + return f"https://{host}/{'/'.join(path_parts)}/did.json" + + +def _fetch_did_json(did_web_base: str) -> dict: + url = _did_web_to_url(did_web_base) + r = requests.get( + url, headers={"Accept": "application/did+json, application/json"}, timeout=15 + ) + r.raise_for_status() + doc = r.json() + if not isinstance(doc, dict): + raise ValueError("did.json must be a JSON object") + return doc + + +def _public_key_from_did_doc(*, kid: str, did_doc: dict) -> bytes: + did_base = kid.split("#", 1)[0].strip() + kid = kid.strip() + vms = did_doc.get("verificationMethod", []) + + def normalize(vm_id: str) -> str: + return did_base + vm_id if vm_id.startswith("#") else vm_id + + for vm in vms: + if not isinstance(vm, dict): + continue + if normalize(vm.get("id", "")) != kid: + continue + pkmb = vm.get("publicKeyMultibase", "") + if not isinstance(pkmb, str) or not pkmb.startswith("z"): + raise ValueError("Invalid publicKeyMultibase") + raw = _decode_multibase_base58btc(pkmb) + if len(raw) == 32: + return raw + code, n = _parse_uvarint(raw) + if code == 0xED and (len(raw) - n) == 32: + return raw[n:] + raise ValueError("publicKeyMultibase is not an Ed25519 key we recognise") + raise ValueError(f"kid not found in did.json: {kid}") + + +def _resolve_public_key_bytes(kid: str) -> bytes: + base = kid.split("#", 1)[0].strip() + if base.startswith("did:key:"): + return _extract_ed25519_pub_from_did_key(kid) + if base.startswith("did:web:"): + return _public_key_from_did_doc(kid=kid, did_doc=_fetch_did_json(base)) + raise ValueError(f"Unsupported DID method: {base}") + + +def sign_proof_jwt(private_key_multibase: str, kid: str, payload_obj: dict) -> str: + seed = _extract_ed25519_seed(private_key_multibase) + private_key = Ed25519PrivateKey.from_private_bytes(seed) + token = jwt.encode( + payload=payload_obj, + key=private_key, + algorithm="EdDSA", + headers={"typ": "JWT", "alg": "EdDSA", "kid": kid}, + ) + return token if isinstance(token, str) else token.decode("utf-8") + + +def verify_proof_jwt(token: str, kid: str) -> tuple[bool, str]: + try: + pub_bytes = _resolve_public_key_bytes(kid) + pub_key = Ed25519PublicKey.from_public_bytes(pub_bytes) + jwt.decode( + token, key=pub_key, algorithms=["EdDSA"], + options={"verify_signature": True, "verify_aud": False, + "verify_exp": False, "verify_iat": False, + "verify_nbf": False, "verify_iss": False}, + ) + return True, "" + except jwt.exceptions.InvalidSignatureError: + return False, "signature verification failed — private key does not match verification method" + except jwt.PyJWTError as e: + return False, f"JWT error: {e}" + except Exception as e: + return False, f"verification error: {e}" + + + + +def extract_user_id_from_token(access_token: str) -> str: + import uuid + try: + payload_part = access_token.split(".")[1] + padding = 4 - len(payload_part) % 4 + payload_part += "=" * (padding % 4) + decoded = json.loads(base64.urlsafe_b64decode(payload_part)) + user_id = decoded.get("sub", "") + if not user_id: + raise ValueError("'sub' claim missing from token") + try: + uuid.UUID(user_id) + except ValueError: + raise ValueError(f"'sub' claim '{user_id}' is not a valid GUID") + return user_id + except ValueError: + raise + except Exception as e: + raise ValueError(f"Failed to extract userId from token: {e}") from e \ No newline at end of file diff --git a/ce/main.py b/ce/main.py new file mode 100644 index 0000000..db488fd --- /dev/null +++ b/ce/main.py @@ -0,0 +1,100 @@ +"""Credential Engine CLI - main entry point.""" + +try: + import truststore + truststore.inject_into_ssl() +except ImportError: + pass + +import click +from rich.console import Console + +from ce.commands.auth import auth_group +from ce.commands.env import env_group +from ce.commands.iir import iir_group +from ce.commands.version import version_command + +console = Console() + +CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"]) + + +@click.group(context_settings=CONTEXT_SETTINGS) +@click.version_option(version="0.1.0", prog_name="ce") +@click.option( + "--debug", + is_flag=True, + default=False, + envvar="CE_DEBUG", + help="Enable debug output (HTTP traces, token refresh events, etc.).", +) +@click.pass_context +def cli(ctx: click.Context, debug: bool) -> None: + """Credential Engine CLI. + + Manage Credential Engine platform resources and services. + + \b + First time? + ce env list Select your environment (dev / sandbox / prod) + ce login Sign in via Keycloak device authorization + ce whoami Confirm who you are signed in as + + \b + Everyday use: + ce logout Sign out + ce env use Switch environment + + \b + IIR — bulk DID upload: + ce iir bulk-upload-dids sign-challenges --csv input.csv + ce iir bulk-upload-dids publish --output-file Output-input-....csv + + \b + Run any command with --help for full options: + ce login --help + ce iir bulk-upload-dids sign-challenges --help + """ + ctx.ensure_object(dict) + ctx.obj["debug"] = debug + + if debug: + from ce.utils.logging import enable_debug + enable_debug() + + +# Top-level commands + +@cli.command("login") +@click.option("--env", "env_name", default=None, metavar="NAME", + help="Environment to log in to (default: active environment).") +@click.pass_context +def login_alias(ctx: click.Context, env_name: str | None) -> None: + """Sign in to Credential Engine via Keycloak device authorization.""" + from ce.commands.auth import login as _login + ctx.invoke(_login, env_name=env_name) + + +@cli.command("logout") +@click.option("--yes", "-y", is_flag=True, help="Skip confirmation prompt.") +@click.pass_context +def logout_alias(ctx: click.Context, yes: bool) -> None: + """Sign out and remove stored credentials.""" + from ce.commands.auth import logout as _logout + ctx.invoke(_logout, yes=yes) + + +@cli.command("whoami") +@click.pass_context +def whoami(ctx: click.Context) -> None: + """Show the currently signed-in account.""" + from ce.commands.auth import account_show as _show + ctx.invoke(_show) + + +# Command groups + +cli.add_command(auth_group) # ce account ... +cli.add_command(env_group) # ce env ... +cli.add_command(iir_group) # ce iir ... +cli.add_command(version_command) # ce version \ No newline at end of file diff --git a/ce/utils/__init__.py b/ce/utils/__init__.py new file mode 100644 index 0000000..45926df --- /dev/null +++ b/ce/utils/__init__.py @@ -0,0 +1,15 @@ +"""Shared utilities for the CE CLI.""" + +from ce.utils.errors import handle_api_errors +from ce.utils.http import APIError, CEHTTPClient, api_client +from ce.utils.logging import enable_debug, get_logger, is_debug + +__all__ = [ + "APIError", + "CEHTTPClient", + "api_client", + "enable_debug", + "get_logger", + "handle_api_errors", + "is_debug", +] \ No newline at end of file diff --git a/ce/utils/errors.py b/ce/utils/errors.py new file mode 100644 index 0000000..a533e63 --- /dev/null +++ b/ce/utils/errors.py @@ -0,0 +1,46 @@ +"""Shared decorators and error-handling utilities for CE commands.""" + +from __future__ import annotations + +import functools +import sys +from typing import Callable + +import click +from rich.console import Console + +from ce.utils.http import APIError + +err_console = Console(stderr=True) + + +def handle_api_errors(f: Callable) -> Callable: + """Decorator: catch APIError / httpx errors and print a friendly message.""" + + @functools.wraps(f) + def wrapper(*args, **kwargs): + try: + return f(*args, **kwargs) + except APIError as exc: + if exc.status_code == 401: + err_console.print( + "[red]✗[/red] Unauthorized. Your session may have expired — run [bold]ce login[/bold]." + ) + elif exc.status_code == 403: + err_console.print( + f"[red]✗[/red] Permission denied: {exc.message}" + ) + elif exc.status_code == 404: + err_console.print( + f"[red]✗[/red] Not found: {exc.message}" + ) + else: + err_console.print(f"[red]✗[/red] API error: {exc.message}") + sys.exit(1) + except click.UsageError: + raise + except Exception as exc: + err_console.print(f"[red]✗[/red] Unexpected error: {exc}") + sys.exit(1) + + return wrapper diff --git a/ce/utils/http.py b/ce/utils/http.py new file mode 100644 index 0000000..91f2fc1 --- /dev/null +++ b/ce/utils/http.py @@ -0,0 +1,106 @@ + +from __future__ import annotations + +from contextlib import contextmanager +from typing import Any, Generator + +import httpx + +from ce.auth.token_manager import require_login +from ce.utils.logging import get_logger + +log = get_logger(__name__) + + +class APIError(Exception): + def __init__(self, status_code: int, message: str) -> None: + self.status_code = status_code + self.message = message + super().__init__(f"HTTP {status_code}: {message}") + + +class CEHTTPClient: + """Thin wrapper around httpx.Client with auth, debug tracing, and error handling.""" + + def __init__(self, base_url: str, access_token: str, timeout: float = 30.0) -> None: + self._client = httpx.Client( + base_url=base_url, + headers={ + "Authorization": f"Bearer {access_token}", + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "ce-cli/0.1.0", + }, + timeout=timeout, + ) + self._base_url = base_url + + def get(self, path: str, **kwargs: Any) -> httpx.Response: + return self._request("GET", path, **kwargs) + + def post(self, path: str, **kwargs: Any) -> httpx.Response: + return self._request("POST", path, **kwargs) + + def put(self, path: str, **kwargs: Any) -> httpx.Response: + return self._request("PUT", path, **kwargs) + + def patch(self, path: str, **kwargs: Any) -> httpx.Response: + return self._request("PATCH", path, **kwargs) + + def delete(self, path: str, **kwargs: Any) -> httpx.Response: + return self._request("DELETE", path, **kwargs) + + def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response: + url = f"{self._base_url}{path}" + start = log.http_request(method, url) + resp = self._client.request(method, path, **kwargs) + log.http_response(resp.status_code, start) + self._raise_for_status(resp) + return resp + + @staticmethod + def _raise_for_status(resp: httpx.Response) -> None: + if resp.is_success: + return + try: + body = resp.json() + message = ( + body.get("message") + or body.get("error_description") + or body.get("error") + or resp.text + ) + except Exception: + message = resp.text or f"HTTP {resp.status_code}" + raise APIError(resp.status_code, message) + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> "CEHTTPClient": + return self + + def __exit__(self, *_: Any) -> None: + self.close() + + +@contextmanager +def api_client(timeout: float = 30.0) -> Generator[CEHTTPClient, None, None]: + """Context manager that yields an authenticated :class:`CEHTTPClient`. + + Automatically fetches (and refreshes) the access token and resolves + the API base URL from the active environment. + + Raises: + click.UsageError: if not logged in or no environment is configured. + """ + access_token = require_login() + from ce.config.context import require_environment + _, _env = require_environment() + base_url = _env.publisher_base.rstrip('/') + log.debug("api_client ready", base_url=base_url) + client = CEHTTPClient(base_url, access_token, timeout=timeout) + try: + yield client + finally: + client.close() \ No newline at end of file diff --git a/ce/utils/logging.py b/ce/utils/logging.py new file mode 100644 index 0000000..bfb2a45 --- /dev/null +++ b/ce/utils/logging.py @@ -0,0 +1,78 @@ + +from __future__ import annotations + +import os +import sys +import time +from typing import Any + +from rich.console import Console +from rich.text import Text + +_stderr = Console(stderr=True) + +_debug_enabled: bool = bool(os.environ.get("CE_DEBUG", "")) + + +def enable_debug() -> None: + """Turn on debug output for the duration of the process.""" + global _debug_enabled + _debug_enabled = True + + +def is_debug() -> bool: + return _debug_enabled + + +class CELogger: + """Lightweight structured logger that writes to stderr via Rich.""" + + _LEVELS = { + "debug": ("[dim blue]DEBUG[/dim blue]", False), + "info": ("[dim green]INFO [/dim green]", False), + "warning": ("[yellow]WARN [/yellow]", False), + "error": ("[red]ERROR[/red]", True), # always shown + } + + def __init__(self, name: str) -> None: + self._name = name.split(".")[-1] + + def _emit(self, level: str, message: str, **fields: Any) -> None: + label, always = self._LEVELS[level] + if not always and not _debug_enabled: + return + + parts = [label, f"[dim]{self._name}[/dim]", message] + if fields: + kv = " ".join(f"[dim]{k}[/dim]=[cyan]{v}[/cyan]" for k, v in fields.items()) + parts.append(kv) + + _stderr.print(" ".join(parts)) + + def debug(self, message: str, **fields: Any) -> None: + self._emit("debug", message, **fields) + + def info(self, message: str, **fields: Any) -> None: + self._emit("info", message, **fields) + + def warning(self, message: str, **fields: Any) -> None: + self._emit("warning", message, **fields) + + def error(self, message: str, **fields: Any) -> None: + self._emit("error", message, **fields) + + + def http_request(self, method: str, url: str, **extra: Any) -> float: + """Log an outgoing HTTP request; returns start time for latency tracking.""" + self.debug(f"→ {method} {url}", **extra) + return time.monotonic() + + def http_response(self, status: int, start: float, **extra: Any) -> None: + elapsed_ms = int((time.monotonic() - start) * 1000) + colour = "green" if status < 400 else "red" + self.debug(f"← [{colour}]{status}[/{colour}] ({elapsed_ms}ms)", **extra) + + +def get_logger(name: str) -> CELogger: + """Return a :class:`CELogger` for the given module name.""" + return CELogger(name) diff --git a/ce/utils/output.py b/ce/utils/output.py new file mode 100644 index 0000000..a683a78 --- /dev/null +++ b/ce/utils/output.py @@ -0,0 +1,134 @@ +"""Output formatting utilities. + +Mirrors az's --output / -o flag: + table - human-friendly Rich table (default) + json - pretty-printed JSON + yaml - YAML (requires pyyaml) + tsv - tab-separated values, script-friendly + none - suppress output (useful in scripts) +""" + +from __future__ import annotations + +import json +from typing import Any, Sequence + +import click +from rich.console import Console +from rich.table import Table, box + +console = Console() + +OutputFormat = str # "table" | "json" | "yaml" | "tsv" | "none" + +OUTPUT_FORMATS = ("table", "json", "yaml", "tsv", "none") + + +def print_output( + data: Any, + fmt: OutputFormat, + *, + columns: Sequence[str] | None = None, + title: str | None = None, +) -> None: + if fmt == "none": + return + + if fmt == "json": + _print_json(data) + elif fmt == "yaml": + _print_yaml(data) + elif fmt == "tsv": + _print_tsv(data, columns=columns) + else: + _print_table(data, columns=columns, title=title) + + + +def _print_json(data: Any) -> None: + from rich.syntax import Syntax + text = json.dumps(data, indent=2, default=str) + console.print(Syntax(text, "json", theme="ansi_dark", background_color="default")) + + +def _print_yaml(data: Any) -> None: + try: + import yaml # type: ignore + except ImportError: + click.echo(json.dumps(data, indent=2, default=str)) + return + from rich.syntax import Syntax + text = yaml.dump(data, default_flow_style=False, allow_unicode=True) + console.print(Syntax(text, "yaml", theme="ansi_dark", background_color="default")) + + +def _print_tsv(data: Any, columns: Sequence[str] | None) -> None: + rows = _to_rows(data) + if not rows: + return + cols = list(columns) if columns else list(rows[0].keys()) + for row in rows: + click.echo("\t".join(str(row.get(c, "")) for c in cols)) + + +def _print_table( + data: Any, + columns: Sequence[str] | None, + title: str | None, +) -> None: + rows = _to_rows(data) + if not rows: + console.print("[dim]No results.[/dim]") + return + + cols = list(columns) if columns else list(rows[0].keys()) + + table = Table( + title=title, + box=box.SIMPLE_HEAD, + show_lines=False, + header_style="bold", + border_style="dim", + ) + for col in cols: + table.add_column(col.replace("_", " ").title()) + + for row in rows: + table.add_row(*[str(row.get(c, "—")) for c in cols]) + + console.print() + console.print(table) + console.print() + + + +def _to_rows(data: Any) -> list[dict]: + """Normalise data to a list of dicts.""" + if isinstance(data, list): + return [r if isinstance(r, dict) else {"value": r} for r in data] + if isinstance(data, dict): + return [data] + return [{"value": data}] + + + + +def output_option(f): # type: ignore[no-untyped-def] + """Decorator that adds ``-o / --output`` to a Click command.""" + return click.option( + "-o", "--output", + "output_fmt", + type=click.Choice(OUTPUT_FORMATS, case_sensitive=False), + default=None, + show_default=True, + help="Output format.", + envvar="CE_OUTPUT", + )(f) + + +def resolve_output_fmt(output_fmt: str | None) -> OutputFormat: + """Return the effective output format, checking config if not passed.""" + if output_fmt: + return output_fmt + from ce.config.context import load_cli_config + return load_cli_config().output_format or "table" diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..015a9dd --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,30 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "ce-cli" +version = "0.1.0" +description = "Credential Engine CLI - Internal platform management tool" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "click>=8.1", + "httpx>=0.27", + "keyring>=25", + "rich>=13", + "pydantic>=2", + "pydantic-settings>=2", + "requests>=2.31", + "truststore>=0.9", + "base58>=2.1", + "PyJWT>=2.8", + "cryptography>=41", +] + +[project.scripts] +ce = "ce.main:cli" + +[tool.setuptools.packages.find] +where = ["."] +include = ["ce*"] \ No newline at end of file From bd013db3b72761bfa822d039f582a6107f4ed890 Mon Sep 17 00:00:00 2001 From: Sneha Edula Date: Mon, 30 Mar 2026 11:57:01 -0400 Subject: [PATCH 3/3] comment updates --- ce/config/context.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/ce/config/context.py b/ce/config/context.py index eb61189..fe3918b 100644 --- a/ce/config/context.py +++ b/ce/config/context.py @@ -32,16 +32,25 @@ class CLIConfig(BaseModel): DEFAULT_ENVIRONMENTS: dict[str, Environment] = { "dev": Environment( label="Development", - client_id="IIR-TokenGenerator-CLI", + client_id="CE-CLI", device_auth_url="http://localhost:8080/realms/CE_Accounts/protocol/openid-connect/auth/device", token_url="http://localhost:8080/realms/CE_Accounts/protocol/openid-connect/token", scopes="openid profile", publisher_base="https://localhost:44330", ssl_verify=False, ), + "test": Environment( + label="Test", + client_id="CE-CLI", + device_auth_url="https://login.test.credentialengine.org/realms/CE-Test/protocol/openid-connect/auth/device", + token_url="https://login.test.credentialengine.org/realms/CE-Test/protocol/openid-connect/token", + scopes="openid profile", + publisher_base="https://sandbox.credentialengine.org/publisher", + ssl_verify=True, + ), "sandbox": Environment( label="Sandbox", - client_id="IIR-TokenGenerator-CLI", + client_id="CE-CLI", device_auth_url="https://login.sandbox.credentialengine.org/realms/CE-Sandbox/protocol/openid-connect/auth/device", token_url="https://login.sandbox.credentialengine.org/realms/CE-Sandbox/protocol/openid-connect/token", scopes="openid profile", @@ -50,7 +59,7 @@ class CLIConfig(BaseModel): ), "prod": Environment( label="Production", - client_id="IIR-TokenGenerator-CLI", + client_id="CE-CLI", device_auth_url="https://login.credentialengine.org/realms/CE-Prod/protocol/openid-connect/auth/device", token_url="https://login.credentialengine.org/realms/CE-Prod/protocol/openid-connect/token", scopes="openid profile",