feat: add authentication commands

- login <username> - login and save token
- register <username> - create account
- logout - clear saved token
- whoami - show current user
- run command now requires auth
- Token stored in ~/.artdag/token.json

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
gilesb
2026-01-07 14:44:16 +00:00
parent fc00a2fc88
commit c505e0a33a

156
artdag.py
View File

@@ -15,6 +15,9 @@ import click
import requests
DEFAULT_SERVER = os.environ.get("ARTDAG_SERVER", "http://localhost:8100")
DEFAULT_L2_SERVER = os.environ.get("ARTDAG_L2", "http://localhost:8200")
CONFIG_DIR = Path.home() / ".artdag"
TOKEN_FILE = CONFIG_DIR / "token.json"
def get_server():
@@ -22,16 +25,54 @@ def get_server():
return DEFAULT_SERVER
def api_get(path: str):
def get_l2_server():
"""Get L2 server URL."""
return DEFAULT_L2_SERVER
def load_token() -> dict:
"""Load saved token from config."""
if TOKEN_FILE.exists():
with open(TOKEN_FILE) as f:
return json.load(f)
return {}
def save_token(token_data: dict):
"""Save token to config."""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
with open(TOKEN_FILE, "w") as f:
json.dump(token_data, f, indent=2)
TOKEN_FILE.chmod(0o600)
def clear_token():
"""Clear saved token."""
if TOKEN_FILE.exists():
TOKEN_FILE.unlink()
def get_auth_header() -> dict:
"""Get Authorization header if token exists."""
token_data = load_token()
token = token_data.get("access_token")
if token:
return {"Authorization": f"Bearer {token}"}
return {}
def api_get(path: str, auth: bool = False):
"""GET request to server."""
resp = requests.get(f"{get_server()}{path}")
headers = get_auth_header() if auth else {}
resp = requests.get(f"{get_server()}{path}", headers=headers)
resp.raise_for_status()
return resp.json()
def api_post(path: str, data: dict = None, params: dict = None):
def api_post(path: str, data: dict = None, params: dict = None, auth: bool = False):
"""POST request to server."""
resp = requests.post(f"{get_server()}{path}", json=data, params=params)
headers = get_auth_header() if auth else {}
resp = requests.post(f"{get_server()}{path}", json=data, params=params, headers=headers)
resp.raise_for_status()
return resp.json()
@@ -39,15 +80,102 @@ def api_post(path: str, data: dict = None, params: dict = None):
@click.group()
@click.option("--server", "-s", envvar="ARTDAG_SERVER", default=DEFAULT_SERVER,
help="L1 server URL")
@click.option("--l2", envvar="ARTDAG_L2", default=DEFAULT_L2_SERVER,
help="L2 server URL")
@click.pass_context
def cli(ctx, server):
def cli(ctx, server, l2):
"""Art DAG Client - interact with L1 rendering server."""
ctx.ensure_object(dict)
ctx.obj["server"] = server
global DEFAULT_SERVER
ctx.obj["l2"] = l2
global DEFAULT_SERVER, DEFAULT_L2_SERVER
DEFAULT_SERVER = server
DEFAULT_L2_SERVER = l2
# ============ Auth Commands ============
@cli.command()
@click.argument("username")
@click.option("--password", "-p", prompt=True, hide_input=True)
def login(username, password):
"""Login to get access token."""
try:
resp = requests.post(
f"{get_l2_server()}/auth/login",
json={"username": username, "password": password}
)
if resp.status_code == 200:
token_data = resp.json()
save_token(token_data)
click.echo(f"Logged in as {token_data['username']}")
click.echo(f"Token expires: {token_data['expires_at']}")
else:
click.echo(f"Login failed: {resp.json().get('detail', 'Unknown error')}", err=True)
sys.exit(1)
except requests.RequestException as e:
click.echo(f"Login failed: {e}", err=True)
sys.exit(1)
@cli.command()
@click.argument("username")
@click.option("--password", "-p", prompt=True, hide_input=True, confirmation_prompt=True)
@click.option("--email", "-e", default=None, help="Email (optional)")
def register(username, password, email):
"""Register a new account."""
try:
resp = requests.post(
f"{get_l2_server()}/auth/register",
json={"username": username, "password": password, "email": email}
)
if resp.status_code == 200:
token_data = resp.json()
save_token(token_data)
click.echo(f"Registered and logged in as {token_data['username']}")
else:
click.echo(f"Registration failed: {resp.json().get('detail', 'Unknown error')}", err=True)
sys.exit(1)
except requests.RequestException as e:
click.echo(f"Registration failed: {e}", err=True)
sys.exit(1)
@cli.command()
def logout():
"""Logout (clear saved token)."""
clear_token()
click.echo("Logged out")
@cli.command()
def whoami():
"""Show current logged-in user."""
token_data = load_token()
if not token_data.get("access_token"):
click.echo("Not logged in")
return
try:
resp = requests.get(
f"{get_l2_server()}/auth/me",
headers={"Authorization": f"Bearer {token_data['access_token']}"}
)
if resp.status_code == 200:
user = resp.json()
click.echo(f"Username: {user['username']}")
click.echo(f"Created: {user['created_at']}")
if user.get('email'):
click.echo(f"Email: {user['email']}")
else:
click.echo("Token invalid or expired. Please login again.", err=True)
clear_token()
except requests.RequestException as e:
click.echo(f"Error: {e}", err=True)
# ============ Server Commands ============
@cli.command()
def info():
"""Show server info."""
@@ -65,11 +193,17 @@ def info():
@click.option("--name", "-n", help="Output name")
@click.option("--wait", "-w", is_flag=True, help="Wait for completion")
def run(recipe, input_hash, name, wait):
"""Start a rendering run.
"""Start a rendering run. Requires login.
RECIPE: Effect/recipe to apply (e.g., dog, identity)
INPUT_HASH: Content hash of input asset
"""
# Check auth
token_data = load_token()
if not token_data.get("access_token"):
click.echo("Not logged in. Please run: artdag login <username>", err=True)
sys.exit(1)
# Resolve named assets
assets = api_get("/assets")
if input_hash in assets:
@@ -83,7 +217,13 @@ def run(recipe, input_hash, name, wait):
if name:
data["output_name"] = name
result = api_post("/runs", data)
try:
result = api_post("/runs", data, auth=True)
except requests.HTTPError as e:
if e.response.status_code == 401:
click.echo("Authentication failed. Please login again.", err=True)
sys.exit(1)
raise
run_id = result["run_id"]
click.echo(f"Run started: {run_id}")