" ["<body>"] forge run <owner/repo> "<command>" [--image alpine] forge cmd <owner/repo> "<command>" [--image alpine] [--port N] forge cmd stop <owner/repo> <run_id> forge 2fa setup forge 2fa verify <code> forge 2fa disable <code> forge ssh-key add "<public key>" ["comment"] forge ssh-key list forge ssh-key rm <id> forge update [--force] forge uninstall [--yes] Config (API base + saved token) lives at ~/.forge/config.json. Only uses the Python standard library — no extra deps to install. """ import os import sys import json import getpass import hashlib import urllib.request import urllib.error import urllib.parse VERSION = '1.1.0' # Cloudflare's bot protection (error 1010) blocks Python's default urllib # User-Agent string outright — send a normal browser-ish one everywhere. USER_AGENT = 'Mozilla/5.0 (compatible; forge-cli/1.1.0)' DEFAULT_API_BASE = os.environ.get('FORGE_API_BASE', 'https://reviewer-shopper-gamma-bag.trycloudflare.com') UPDATE_URL = os.environ.get('FORGE_UPDATE_URL', 'https://update.luisforge.pages.dev/forge.py') CONFIG_DIR = os.path.expanduser('~/.forge') CONFIG_PATH = os.path.join(CONFIG_DIR, 'config.json') def load_config(): if os.path.exists(CONFIG_PATH): with open(CONFIG_PATH) as f: return json.load(f) return {} def save_config(cfg): os.makedirs(CONFIG_DIR, exist_ok=True) with open(CONFIG_PATH, 'w') as f: json.dump(cfg, f, indent=2) os.chmod(CONFIG_PATH, 0o600) # contains an auth token — keep it user-readable only def api_base(): return load_config().get('api_base') or DEFAULT_API_BASE def token(): return load_config().get('token') def request(method, path, body=None, auth=True, raw=False): url = api_base().rstrip('/') + path headers = {'User-Agent': USER_AGENT} data = None if body is not None: headers['Content-Type'] = 'application/json' data = json.dumps(body).encode() if auth and token(): headers['Authorization'] = f'Bearer {token()}' req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=30) as res: content = res.read() return res.status, (content if raw else json.loads(content)) except urllib.error.HTTPError as e: content = e.read() try: return e.code, json.loads(content) except json.JSONDecodeError: return e.code, {'error': content.decode(errors='replace')} except urllib.error.URLError as e: print(f"error: couldn't reach {api_base()} ({e.reason})", file=sys.stderr) sys.exit(1) def die(msg): print(f"error: {msg}", file=sys.stderr) sys.exit(1) # ---------------- auth ---------------- def cmd_auth_signup(args): username = input('Username: ').strip() email = input('Email: ').strip() password = getpass.getpass('Password: ') status, data = request('POST', '/api/account/register', { 'username': username, 'email': email, 'password': password }, auth=False) if status != 200: die(data.get('error', 'registration failed')) cfg = load_config() cfg['token'] = data['token'] cfg['username'] = data['username'] save_config(cfg) print(f"Registered and logged in as {data['username']}.") def cmd_auth_login(args): username = input('Username: ').strip() password = getpass.getpass('Password: ') body = {'username': username, 'password': password} status, data = request('POST', '/api/account/login', body, auth=False) if status != 200 and data.get('require_2fa'): body['totp_code'] = input('2FA code: ').strip() status, data = request('POST', '/api/account/login', body, auth=False) if status != 200: die(data.get('error', 'login failed')) cfg = load_config() cfg['token'] = data['token'] cfg['username'] = data['username'] save_config(cfg) print(f"Logged in as {data['username']}.") def cmd_auth_logout(args): cfg = load_config() cfg.pop('token', None) cfg.pop('username', None) save_config(cfg) print('Logged out.') def cmd_auth_whoami(args): cfg = load_config() if not cfg.get('token'): print('Not logged in.') else: print(f"Logged in as {cfg.get('username', '?')} ({api_base()})") # ---------------- repos ---------------- def cmd_repos(args): status, data = request('GET', '/api/repos', auth=False) if status != 200: die(data.get('error', 'failed to list repos')) if not data['repos']: print('No public repos yet.') return for r in data['repos']: priv = ' [private]' if r.get('private') else '' print(f"{r['owner']}/{r['name']}{priv} — {r.get('description') or 'no description'}") def cmd_create(args): if not args: die('usage: forge create <name> [--private] [--desc "..."]') if not token(): die('not logged in — run `forge auth login` first') name = args[0] private = '--private' in args desc = '' if '--desc' in args: i = args.index('--desc') if i + 1 < len(args): desc = args[i + 1] status, data = request('POST', '/api/repos', {'name': name, 'description': desc, 'private': private}) if status != 200: die(data.get('error', 'failed to create repo')) print(f"Created {data['owner']}/{data['name']}.") def _split_repo(spec): if '/' not in spec: die('expected format: owner/repo') owner, name = spec.split('/', 1) return owner, name def cmd_delete(args): if not args: die('usage: forge delete <owner/repo>') owner, name = _split_repo(args[0]) if not token(): die('not logged in — run `forge auth login` first') confirm = input(f"Type '{name}' to confirm deleting {owner}/{name}: ") if confirm != name: die('confirmation did not match — aborted') status, data = request('DELETE', f'/api/repos/{owner}/{name}') if status != 200: die(data.get('error', 'failed to delete repo')) print(f"Deleted {owner}/{name}.") def cmd_push(args): if len(args) < 2: die('usage: forge push <owner/repo> <local_file> [remote_path]') owner, name = _split_repo(args[0]) local_file = args[1] if not os.path.isfile(local_file): die(f"no such file: {local_file}") remote_path = args[2] if len(args) > 2 else os.path.basename(local_file) if not token(): die('not logged in — run `forge auth login` first') remote_dir = os.path.dirname(remote_path) filename = os.path.basename(remote_path) boundary = '----forgeboundary' with open(local_file, 'rb') as f: file_data = f.read() parts = [] for field, value in (('path', remote_dir), ('message', f'Push {filename} via forge CLI')): parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{field}"\r\n\r\n{value}\r\n'.encode()) parts.append( f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\n' f'Content-Type: application/octet-stream\r\n\r\n'.encode() + file_data + b'\r\n' ) parts.append(f'--{boundary}--\r\n'.encode()) body = b''.join(parts) url = api_base().rstrip('/') + f'/api/repos/{owner}/{name}/upload' req = urllib.request.Request(url, data=body, method='POST', headers={ 'Content-Type': f'multipart/form-data; boundary={boundary}', 'Authorization': f'Bearer {token()}', 'User-Agent': USER_AGENT, }) try: with urllib.request.urlopen(req, timeout=30) as res: data = json.loads(res.read()) except urllib.error.HTTPError as e: die(json.loads(e.read()).get('error', 'push failed')) print(f"Pushed {remote_path} — commit {data['commit'][:8]}.") def cmd_files(args): if not args: die('usage: forge files <owner/repo> [path]') owner, name = _split_repo(args[0]) path = args[1] if len(args) > 1 else '' status, data = request('GET', f'/api/repos/{owner}/{name}/files?path={urllib.parse.quote(path)}', auth=False) if status != 200: die(data.get('error', 'failed to list files')) if data.get('empty'): print('(empty repo)') return for e in data['entries']: print(f"{'d' if e['type'] == 'dir' else '-'} {e['name']}") def cmd_cat(args): if len(args) < 2: die('usage: forge cat <owner/repo> <path>') owner, name = _split_repo(args[0]) status, data = request('GET', f'/api/repos/{owner}/{name}/blob?path={urllib.parse.quote(args[1])}', auth=False) if status != 200: die(data.get('error', 'failed to read file')) print(data['content'], end='') def cmd_log(args): if not args: die('usage: forge log <owner/repo>') owner, name = _split_repo(args[0]) status, data = request('GET', f'/api/repos/{owner}/{name}/commits', auth=False) if status != 200: die(data.get('error', 'failed to fetch commits')) if not data['commits']: print('(no commits yet)') return for c in data['commits']: print(f"{c['sha'][:8]} {c['message'].splitlines()[0]}") def cmd_issues(args): if not args or args[0] not in ('list', 'create'): die('usage: forge issues list <owner/repo>\n forge issues create <owner/repo> "<title>" ["<body>"]') sub, rest = args[0], args[1:] if not rest: die('usage: forge issues list <owner/repo>') owner, name = _split_repo(rest[0]) if sub == 'list': status, data = request('GET', f'/api/repos/{owner}/{name}/issues', auth=False) if status != 200: die(data.get('error', 'failed to list issues')) if not data['issues']: print('(no issues yet)') return for i in data['issues']: print(f"#{i['number']} [{i['status']}] {i['title']} (by {i['author']})") else: if len(rest) < 2: die('usage: forge issues create <owner/repo> "<title>" ["<body>"]') if not token(): die('not logged in — run `forge auth login` first') title = rest[1] body = rest[2] if len(rest) > 2 else '' status, data = request('POST', f'/api/repos/{owner}/{name}/issues', {'title': title, 'body': body}) if status != 200: die(data.get('error', 'failed to create issue')) print(f"Opened issue #{data['number']}.") def cmd_run(args): if len(args) < 2: die('usage: forge run <owner/repo> "<command>" [--image alpine]') owner, name = _split_repo(args[0]) command = args[1] image = 'alpine' if '--image' in args: i = args.index('--image') if i + 1 < len(args): image = args[i + 1] if not token(): die('not logged in — run `forge auth login` first') status, data = request('POST', f'/api/repos/{owner}/{name}/cmd/run', {'image': image, 'cmd': command}) if status != 200: die(data.get('error', 'failed to start run')) run_id = data['run_id'] print(f"Started run {run_id[:8]} — polling...") import time as _time for _ in range(60): _time.sleep(2) status, data = request('GET', f'/api/repos/{owner}/{name}/actions', auth=False) run = next((r for r in data.get('runs', []) if r['id'] == run_id), None) if run and run['status'] != 'running': print(f"--- {run['status']} ---") print(run.get('log', '')) return print('(still running — check the web UI for the result)') def cmd_cmd(args): if args and args[0] == 'stop': if len(args) < 3: die('usage: forge cmd stop <owner/repo> <run_id>') owner, name = _split_repo(args[1]) run_id = args[2] if not token(): die('not logged in — run `forge auth login` first') status, data = request('POST', f'/api/repos/{owner}/{name}/server/{run_id}/stop') if status != 200: die(data.get('error', 'failed to stop')) print('Stopped.') return if len(args) < 2: die('usage: forge cmd <owner/repo> "<command>" [--image alpine] [--port N]') owner, name = _split_repo(args[0]) command = args[1] image = 'alpine' if '--image' in args: i = args.index('--image') if i + 1 < len(args): image = args[i + 1] port = None if '--port' in args: i = args.index('--port') if i + 1 < len(args): port = int(args[i + 1]) if not token(): die('not logged in — run `forge auth login` first') import time as _time if port is not None: # long-running "host it publicly with bore" mode status, data = request('POST', f'/api/repos/{owner}/{name}/server/run', {'image': image, 'cmd': command, 'port': port}) if status != 200: die(data.get('error', 'failed to start server')) run_id = data['run_id'] print(f"Started server run {run_id[:8]} — waiting for the public address...") for _ in range(30): _time.sleep(2) status, data = request('GET', f'/api/repos/{owner}/{name}/actions', auth=False) run = next((r for r in data.get('runs', []) if r['id'] == run_id), None) if run and run.get('public_addr'): print(f"Public address: {run['public_addr']}") print(f"Stop with: forge cmd stop {owner}/{name} {run_id}") return if run and run['status'] == 'failed': die(run.get('log', 'server failed to start')) print('(still starting — check the web UI, or the address may just be slow to register)') print(f"Run id: {run_id} — stop with: forge cmd stop {owner}/{name} {run_id}") return # one-shot mode — same behavior as `forge run` status, data = request('POST', f'/api/repos/{owner}/{name}/cmd/run', {'image': image, 'cmd': command}) if status != 200: die(data.get('error', 'failed to start run')) run_id = data['run_id'] print(f"Started run {run_id[:8]} — polling...") for _ in range(60): _time.sleep(2) status, data = request('GET', f'/api/repos/{owner}/{name}/actions', auth=False) run = next((r for r in data.get('runs', []) if r['id'] == run_id), None) if run and run['status'] != 'running': print(f"--- {run['status']} ---") print(run.get('log', '')) return print('(still running — check the web UI for the result)') def cmd_ssh_key(args): if not args or args[0] not in ('add', 'list', 'rm'): die('usage: forge ssh-key add "<public key>" ["comment"]\n' ' forge ssh-key list\n' ' forge ssh-key rm <id>') if not token(): die('not logged in — run `forge auth login` first') if args[0] == 'add': if len(args) < 2: die('usage: forge ssh-key add "<public key>" ["comment"]') public_key = args[1] comment = args[2] if len(args) > 2 else '' status, data = request('POST', '/api/account/ssh-keys', {'public_key': public_key, 'comment': comment}) if status != 200: die(data.get('error', 'failed to add key')) print(f"Added key {data['id'][:8]}.") elif args[0] == 'list': status, data = request('GET', '/api/account/ssh-keys') if status != 200: die(data.get('error', 'failed to list keys')) if not data['keys']: print('(no SSH keys registered)') return for k in data['keys']: print(f"{k['id'][:8]} {k['type']} {k['preview']} {k['comment']} added {k['created_at']}") else: if len(args) < 2: die('usage: forge ssh-key rm <id>') status, data = request('DELETE', f'/api/account/ssh-keys/{args[1]}') if status != 200: die(data.get('error', 'failed to remove key')) print('Removed.') def cmd_2fa(args): if not args or args[0] not in ('setup', 'verify', 'disable'): die('usage: forge 2fa setup\n forge 2fa verify <code>\n forge 2fa disable <code>') if not token(): die('not logged in — run `forge auth login` first') if args[0] == 'setup': status, data = request('POST', '/api/account/2fa/setup') if status != 200: die(data.get('error', 'failed to start 2FA setup')) print(f"Secret: {data['secret']}") print(f"Add manually, or use this URI in your authenticator app:") print(data['otpauth_url']) print("Then run: forge 2fa verify <code>") elif args[0] == 'verify': if len(args) < 2: die('usage: forge 2fa verify <code>') status, data = request('POST', '/api/account/2fa/verify', {'code': args[1]}) if status != 200: die(data.get('error', 'verification failed')) print('2FA enabled.') else: if len(args) < 2: die('usage: forge 2fa disable <code>') status, data = request('POST', '/api/account/2fa/disable', {'code': args[1]}) if status != 200: die(data.get('error', 'failed to disable 2FA')) print('2FA disabled.') # ---------------- clone ---------------- def fetch_dir(owner, name, remote_path, local_dir): status, data = request('GET', f'/api/repos/{owner}/{name}/files?path={urllib.parse.quote(remote_path)}', auth=False) if status != 200: die(data.get('error', f'failed to list {remote_path or "/"}')) if data.get('empty'): return 0 count = 0 for entry in data['entries']: remote_child = f"{remote_path}/{entry['name']}" if remote_path else entry['name'] local_child = os.path.join(local_dir, entry['name']) if entry['type'] == 'dir': os.makedirs(local_child, exist_ok=True) count += fetch_dir(owner, name, remote_child, local_child) else: bstatus, bdata = request('GET', f'/api/repos/{owner}/{name}/blob?path={urllib.parse.quote(remote_child)}', auth=False) if bstatus != 200: print(f"warning: couldn't fetch {remote_child}: {bdata.get('error')}", file=sys.stderr) continue with open(local_child, 'w', encoding='utf-8') as f: f.write(bdata['content']) count += 1 return count def cmd_clone(args): if not args: die('usage: forge clone <owner/repo> [dest]') spec = args[0] # accept a bare "owner/repo", or a full LuisForge clone URL if '/' in spec and spec.startswith(('http://', 'https://')): path = urllib.parse.urlparse(spec).path # /git/<owner>/<repo>.git parts = [p for p in path.split('/') if p] if len(parts) < 3: die(f'could not parse repo from URL: {spec}') owner, name = parts[-2], parts[-1].removesuffix('.git') else: if '/' not in spec: die('expected format: owner/repo') owner, name = spec.split('/', 1) dest = args[1] if len(args) > 1 else name if os.path.exists(dest) and os.listdir(dest): die(f"'{dest}' already exists and isn't empty") os.makedirs(dest, exist_ok=True) status, meta = request('GET', f'/api/repos/{owner}/{name}', auth=False) if status != 200: die(meta.get('error', 'repo not found')) print(f"Cloning {owner}/{name} into '{dest}'...") count = fetch_dir(owner, name, '', dest) print(f"Done — {count} file(s) written.") # ---------------- self-update ---------------- def cmd_update(args): force = '--force' in args this_path = os.path.realpath(__file__) try: req = urllib.request.Request(UPDATE_URL, headers={'User-Agent': USER_AGENT}) with urllib.request.urlopen(req, timeout=10) as resp: remote_bytes = resp.read() except urllib.error.URLError as e: die(f"couldn't reach {UPDATE_URL} ({e.reason})") with open(this_path, 'rb') as f: local_bytes = f.read() if not force and hashlib.sha256(remote_bytes).digest() == hashlib.sha256(local_bytes).digest(): print(f'Already up to date (v{VERSION}).') return tmp_path = this_path + '.new' with open(tmp_path, 'wb') as f: f.write(remote_bytes) os.chmod(tmp_path, 0o755) os.replace(tmp_path, this_path) print(f'Updated {this_path} from {UPDATE_URL}.') print("Run 'forge -h' to confirm it's the version you expect.") # ---------------- uninstall ---------------- PATH_MARKER = '# added by LuisForge install.sh' def _strip_installer_path_line(rc_path): if not os.path.exists(rc_path): return False with open(rc_path) as f: lines = f.readlines() out = [] changed = False skip_next_blank = False i = 0 while i < len(lines): line = lines[i] if line.strip() == PATH_MARKER: changed = True i += 1 # also drop the export line right after the marker, and one # leading blank line the installer adds before the marker if i < len(lines) and 'export PATH=' in lines[i]: i += 1 if out and out[-1].strip() == '': out.pop() continue out.append(line) i += 1 if changed: with open(rc_path, 'w') as f: f.writelines(out) return changed def cmd_uninstall(args): this_path = os.path.realpath(__file__) codey_path = os.path.join(os.path.dirname(this_path), 'codey') has_codey = os.path.exists(codey_path) if '--yes' not in args and '-y' not in args: extra = f' and {codey_path}' if has_codey else '' answer = input(f"Remove {this_path}{extra} and ~/.forge (config + saved token)? [y/N] ").strip().lower() if answer not in ('y', 'yes'): print('Aborted.') return if os.path.isdir(CONFIG_DIR): import shutil shutil.rmtree(CONFIG_DIR) print(f'Removed {CONFIG_DIR}') for rc in (os.path.expanduser('~/.zshrc'), os.path.expanduser('~/.bashrc'), os.path.expanduser('~/.profile')): if _strip_installer_path_line(rc): print(f'Removed PATH entry from {rc}') if has_codey: print(f'Removing {codey_path}') os.remove(codey_path) print(f'Removing {this_path}') os.remove(this_path) print('forge has been uninstalled.') # ---------------- dispatch ---------------- COMMANDS = { ('auth', 'signup'): cmd_auth_signup, ('auth', 'login'): cmd_auth_login, ('auth', 'logout'): cmd_auth_logout, ('auth', 'whoami'): cmd_auth_whoami, } TOP_LEVEL = { 'clone': cmd_clone, 'repos': cmd_repos, 'create': cmd_create, 'delete': cmd_delete, 'push': cmd_push, 'files': cmd_files, 'cat': cmd_cat, 'log': cmd_log, 'issues': cmd_issues, 'run': cmd_run, 'cmd': cmd_cmd, '2fa': cmd_2fa, 'ssh-key': cmd_ssh_key, 'update': cmd_update, 'uninstall': cmd_uninstall, } def main(): argv = sys.argv[1:] if not argv or argv[0] in ('-h', '--help'): print(__doc__) return if argv[0] == 'auth': if len(argv) < 2 or (argv[0], argv[1]) not in COMMANDS: print("usage: forge auth <signup|login|logout|whoami>", file=sys.stderr) sys.exit(1) COMMANDS[(argv[0], argv[1])](argv[2:]) return if argv[0] in TOP_LEVEL: TOP_LEVEL[argv[0]](argv[1:]) return print(f"unknown command: {argv[0]}", file=sys.stderr) print(__doc__) sys.exit(1) if __name__ == '__main_

#!/usr/bin/env python3 """ forge — a tiny CLI client for LuisForge. Usage: forge auth signup forge auth login forge auth logout forge auth whoami forge clone [dest] forge repos forge create [--private] [--desc "..."] forge delete forge push [remote_path] forge files [path] forge cat forge log forge issues list forge issues create "" ["<body>"] forge run <owner/repo> "<command>" [--image alpine] forge cmd <owner/repo> "<command>" [--image alpine] [--port N] forge cmd stop <owner/repo> <run_id> forge 2fa setup forge 2fa verify <code> forge 2fa disable <code> forge ssh-key add "<public key>" ["comment"] forge ssh-key list forge ssh-key rm <id> forge update [--force] forge uninstall [--yes] Config (API base + saved token) lives at ~/.forge/config.json. Only uses the Python standard library — no extra deps to install. """ import os import sys import json import getpass import hashlib import urllib.request import urllib.error import urllib.parse VERSION = '1.1.0' # Cloudflare's bot protection (error 1010) blocks Python's default urllib # User-Agent string outright — send a normal browser-ish one everywhere. USER_AGENT = 'Mozilla/5.0 (compatible; forge-cli/1.1.0)' DEFAULT_API_BASE = os.environ.get('FORGE_API_BASE', 'https://reviewer-shopper-gamma-bag.trycloudflare.com') UPDATE_URL = os.environ.get('FORGE_UPDATE_URL', 'https://update.luisforge.pages.dev/forge.py') CONFIG_DIR = os.path.expanduser('~/.forge') CONFIG_PATH = os.path.join(CONFIG_DIR, 'config.json') def load_config(): if os.path.exists(CONFIG_PATH): with open(CONFIG_PATH) as f: return json.load(f) return {} def save_config(cfg): os.makedirs(CONFIG_DIR, exist_ok=True) with open(CONFIG_PATH, 'w') as f: json.dump(cfg, f, indent=2) os.chmod(CONFIG_PATH, 0o600) # contains an auth token — keep it user-readable only def api_base(): return load_config().get('api_base') or DEFAULT_API_BASE def token(): return load_config().get('token') def request(method, path, body=None, auth=True, raw=False): url = api_base().rstrip('/') + path headers = {'User-Agent': USER_AGENT} data = None if body is not None: headers['Content-Type'] = 'application/json' data = json.dumps(body).encode() if auth and token(): headers['Authorization'] = f'Bearer {token()}' req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=30) as res: content = res.read() return res.status, (content if raw else json.loads(content)) except urllib.error.HTTPError as e: content = e.read() try: return e.code, json.loads(content) except json.JSONDecodeError: return e.code, {'error': content.decode(errors='replace')} except urllib.error.URLError as e: print(f"error: couldn't reach {api_base()} ({e.reason})", file=sys.stderr) sys.exit(1) def die(msg): print(f"error: {msg}", file=sys.stderr) sys.exit(1) # ---------------- auth ---------------- def cmd_auth_signup(args): username = input('Username: ').strip() email = input('Email: ').strip() password = getpass.getpass('Password: ') status, data = request('POST', '/api/account/register', { 'username': username, 'email': email, 'password': password }, auth=False) if status != 200: die(data.get('error', 'registration failed')) cfg = load_config() cfg['token'] = data['token'] cfg['username'] = data['username'] save_config(cfg) print(f"Registered and logged in as {data['username']}.") def cmd_auth_login(args): username = input('Username: ').strip() password = getpass.getpass('Password: ') body = {'username': username, 'password': password} status, data = request('POST', '/api/account/login', body, auth=False) if status != 200 and data.get('require_2fa'): body['totp_code'] = input('2FA code: ').strip() status, data = request('POST', '/api/account/login', body, auth=False) if status != 200: die(data.get('error', 'login failed')) cfg = load_config() cfg['token'] = data['token'] cfg['username'] = data['username'] save_config(cfg) print(f"Logged in as {data['username']}.") def cmd_auth_logout(args): cfg = load_config() cfg.pop('token', None) cfg.pop('username', None) save_config(cfg) print('Logged out.') def cmd_auth_whoami(args): cfg = load_config() if not cfg.get('token'): print('Not logged in.') else: print(f"Logged in as {cfg.get('username', '?')} ({api_base()})") # ---------------- repos ---------------- def cmd_repos(args): status, data = request('GET', '/api/repos', auth=False) if status != 200: die(data.get('error', 'failed to list repos')) if not data['repos']: print('No public repos yet.') return for r in data['repos']: priv = ' [private]' if r.get('private') else '' print(f"{r['owner']}/{r['name']}{priv} — {r.get('description') or 'no description'}") def cmd_create(args): if not args: die('usage: forge create <name> [--private] [--desc "..."]') if not token(): die('not logged in — run `forge auth login` first') name = args[0] private = '--private' in args desc = '' if '--desc' in args: i = args.index('--desc') if i + 1 < len(args): desc = args[i + 1] status, data = request('POST', '/api/repos', {'name': name, 'description': desc, 'private': private}) if status != 200: die(data.get('error', 'failed to create repo')) print(f"Created {data['owner']}/{data['name']}.") def _split_repo(spec): if '/' not in spec: die('expected format: owner/repo') owner, name = spec.split('/', 1) return owner, name def cmd_delete(args): if not args: die('usage: forge delete <owner/repo>') owner, name = _split_repo(args[0]) if not token(): die('not logged in — run `forge auth login` first') confirm = input(f"Type '{name}' to confirm deleting {owner}/{name}: ") if confirm != name: die('confirmation did not match — aborted') status, data = request('DELETE', f'/api/repos/{owner}/{name}') if status != 200: die(data.get('error', 'failed to delete repo')) print(f"Deleted {owner}/{name}.") def cmd_push(args): if len(args) < 2: die('usage: forge push <owner/repo> <local_file> [remote_path]') owner, name = _split_repo(args[0]) local_file = args[1] if not os.path.isfile(local_file): die(f"no such file: {local_file}") remote_path = args[2] if len(args) > 2 else os.path.basename(local_file) if not token(): die('not logged in — run `forge auth login` first') remote_dir = os.path.dirname(remote_path) filename = os.path.basename(remote_path) boundary = '----forgeboundary' with open(local_file, 'rb') as f: file_data = f.read() parts = [] for field, value in (('path', remote_dir), ('message', f'Push {filename} via forge CLI')): parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{field}"\r\n\r\n{value}\r\n'.encode()) parts.append( f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\n' f'Content-Type: application/octet-stream\r\n\r\n'.encode() + file_data + b'\r\n' ) parts.append(f'--{boundary}--\r\n'.encode()) body = b''.join(parts) url = api_base().rstrip('/') + f'/api/repos/{owner}/{name}/upload' req = urllib.request.Request(url, data=body, method='POST', headers={ 'Content-Type': f'multipart/form-data; boundary={boundary}', 'Authorization': f'Bearer {token()}', 'User-Agent': USER_AGENT, }) try: with urllib.request.urlopen(req, timeout=30) as res: data = json.loads(res.read()) except urllib.error.HTTPError as e: die(json.loads(e.read()).get('error', 'push failed')) print(f"Pushed {remote_path} — commit {data['commit'][:8]}.") def cmd_files(args): if not args: die('usage: forge files <owner/repo> [path]') owner, name = _split_repo(args[0]) path = args[1] if len(args) > 1 else '' status, data = request('GET', f'/api/repos/{owner}/{name}/files?path={urllib.parse.quote(path)}', auth=False) if status != 200: die(data.get('error', 'failed to list files')) if data.get('empty'): print('(empty repo)') return for e in data['entries']: print(f"{'d' if e['type'] == 'dir' else '-'} {e['name']}") def cmd_cat(args): if len(args) < 2: die('usage: forge cat <owner/repo> <path>') owner, name = _split_repo(args[0]) status, data = request('GET', f'/api/repos/{owner}/{name}/blob?path={urllib.parse.quote(args[1])}', auth=False) if status != 200: die(data.get('error', 'failed to read file')) print(data['content'], end='') def cmd_log(args): if not args: die('usage: forge log <owner/repo>') owner, name = _split_repo(args[0]) status, data = request('GET', f'/api/repos/{owner}/{name}/commits', auth=False) if status != 200: die(data.get('error', 'failed to fetch commits')) if not data['commits']: print('(no commits yet)') return for c in data['commits']: print(f"{c['sha'][:8]} {c['message'].splitlines()[0]}") def cmd_issues(args): if not args or args[0] not in ('list', 'create'): die('usage: forge issues list <owner/repo>\n forge issues create <owner/repo> "<title>" ["<body>"]') sub, rest = args[0], args[1:] if not rest: die('usage: forge issues list <owner/repo>') owner, name = _split_repo(rest[0]) if sub == 'list': status, data = request('GET', f'/api/repos/{owner}/{name}/issues', auth=False) if status != 200: die(data.get('error', 'failed to list issues')) if not data['issues']: print('(no issues yet)') return for i in data['issues']: print(f"#{i['number']} [{i['status']}] {i['title']} (by {i['author']})") else: if len(rest) < 2: die('usage: forge issues create <owner/repo> "<title>" ["<body>"]') if not token(): die('not logged in — run `forge auth login` first') title = rest[1] body = rest[2] if len(rest) > 2 else '' status, data = request('POST', f'/api/repos/{owner}/{name}/issues', {'title': title, 'body': body}) if status != 200: die(data.get('error', 'failed to create issue')) print(f"Opened issue #{data['number']}.") def cmd_run(args): if len(args) < 2: die('usage: forge run <owner/repo> "<command>" [--image alpine]') owner, name = _split_repo(args[0]) command = args[1] image = 'alpine' if '--image' in args: i = args.index('--image') if i + 1 < len(args): image = args[i + 1] if not token(): die('not logged in — run `forge auth login` first') status, data = request('POST', f'/api/repos/{owner}/{name}/cmd/run', {'image': image, 'cmd': command}) if status != 200: die(data.get('error', 'failed to start run')) run_id = data['run_id'] print(f"Started run {run_id[:8]} — polling...") import time as _time for _ in range(60): _time.sleep(2) status, data = request('GET', f'/api/repos/{owner}/{name}/actions', auth=False) run = next((r for r in data.get('runs', []) if r['id'] == run_id), None) if run and run['status'] != 'running': print(f"--- {run['status']} ---") print(run.get('log', '')) return print('(still running — check the web UI for the result)') def cmd_cmd(args): if args and args[0] == 'stop': if len(args) < 3: die('usage: forge cmd stop <owner/repo> <run_id>') owner, name = _split_repo(args[1]) run_id = args[2] if not token(): die('not logged in — run `forge auth login` first') status, data = request('POST', f'/api/repos/{owner}/{name}/server/{run_id}/stop') if status != 200: die(data.get('error', 'failed to stop')) print('Stopped.') return if len(args) < 2: die('usage: forge cmd <owner/repo> "<command>" [--image alpine] [--port N]') owner, name = _split_repo(args[0]) command = args[1] image = 'alpine' if '--image' in args: i = args.index('--image') if i + 1 < len(args): image = args[i + 1] port = None if '--port' in args: i = args.index('--port') if i + 1 < len(args): port = int(args[i + 1]) if not token(): die('not logged in — run `forge auth login` first') import time as _time if port is not None: # long-running "host it publicly with bore" mode status, data = request('POST', f'/api/repos/{owner}/{name}/server/run', {'image': image, 'cmd': command, 'port': port}) if status != 200: die(data.get('error', 'failed to start server')) run_id = data['run_id'] print(f"Started server run {run_id[:8]} — waiting for the public address...") for _ in range(30): _time.sleep(2) status, data = request('GET', f'/api/repos/{owner}/{name}/actions', auth=False) run = next((r for r in data.get('runs', []) if r['id'] == run_id), None) if run and run.get('public_addr'): print(f"Public address: {run['public_addr']}") print(f"Stop with: forge cmd stop {owner}/{name} {run_id}") return if run and run['status'] == 'failed': die(run.get('log', 'server failed to start')) print('(still starting — check the web UI, or the address may just be slow to register)') print(f"Run id: {run_id} — stop with: forge cmd stop {owner}/{name} {run_id}") return # one-shot mode — same behavior as `forge run` status, data = request('POST', f'/api/repos/{owner}/{name}/cmd/run', {'image': image, 'cmd': command}) if status != 200: die(data.get('error', 'failed to start run')) run_id = data['run_id'] print(f"Started run {run_id[:8]} — polling...") for _ in range(60): _time.sleep(2) status, data = request('GET', f'/api/repos/{owner}/{name}/actions', auth=False) run = next((r for r in data.get('runs', []) if r['id'] == run_id), None) if run and run['status'] != 'running': print(f"--- {run['status']} ---") print(run.get('log', '')) return print('(still running — check the web UI for the result)') def cmd_ssh_key(args): if not args or args[0] not in ('add', 'list', 'rm'): die('usage: forge ssh-key add "<public key>" ["comment"]\n' ' forge ssh-key list\n' ' forge ssh-key rm <id>') if not token(): die('not logged in — run `forge auth login` first') if args[0] == 'add': if len(args) < 2: die('usage: forge ssh-key add "<public key>" ["comment"]') public_key = args[1] comment = args[2] if len(args) > 2 else '' status, data = request('POST', '/api/account/ssh-keys', {'public_key': public_key, 'comment': comment}) if status != 200: die(data.get('error', 'failed to add key')) print(f"Added key {data['id'][:8]}.") elif args[0] == 'list': status, data = request('GET', '/api/account/ssh-keys') if status != 200: die(data.get('error', 'failed to list keys')) if not data['keys']: print('(no SSH keys registered)') return for k in data['keys']: print(f"{k['id'][:8]} {k['type']} {k['preview']} {k['comment']} added {k['created_at']}") else: if len(args) < 2: die('usage: forge ssh-key rm <id>') status, data = request('DELETE', f'/api/account/ssh-keys/{args[1]}') if status != 200: die(data.get('error', 'failed to remove key')) print('Removed.') def cmd_2fa(args): if not args or args[0] not in ('setup', 'verify', 'disable'): die('usage: forge 2fa setup\n forge 2fa verify <code>\n forge 2fa disable <code>') if not token(): die('not logged in — run `forge auth login` first') if args[0] == 'setup': status, data = request('POST', '/api/account/2fa/setup') if status != 200: die(data.get('error', 'failed to start 2FA setup')) print(f"Secret: {data['secret']}") print(f"Add manually, or use this URI in your authenticator app:") print(data['otpauth_url']) print("Then run: forge 2fa verify <code>") elif args[0] == 'verify': if len(args) < 2: die('usage: forge 2fa verify <code>') status, data = request('POST', '/api/account/2fa/verify', {'code': args[1]}) if status != 200: die(data.get('error', 'verification failed')) print('2FA enabled.') else: if len(args) < 2: die('usage: forge 2fa disable <code>') status, data = request('POST', '/api/account/2fa/disable', {'code': args[1]}) if status != 200: die(data.get('error', 'failed to disable 2FA')) print('2FA disabled.') # ---------------- clone ---------------- def fetch_dir(owner, name, remote_path, local_dir): status, data = request('GET', f'/api/repos/{owner}/{name}/files?path={urllib.parse.quote(remote_path)}', auth=False) if status != 200: die(data.get('error', f'failed to list {remote_path or "/"}')) if data.get('empty'): return 0 count = 0 for entry in data['entries']: remote_child = f"{remote_path}/{entry['name']}" if remote_path else entry['name'] local_child = os.path.join(local_dir, entry['name']) if entry['type'] == 'dir': os.makedirs(local_child, exist_ok=True) count += fetch_dir(owner, name, remote_child, local_child) else: bstatus, bdata = request('GET', f'/api/repos/{owner}/{name}/blob?path={urllib.parse.quote(remote_child)}', auth=False) if bstatus != 200: print(f"warning: couldn't fetch {remote_child}: {bdata.get('error')}", file=sys.stderr) continue with open(local_child, 'w', encoding='utf-8') as f: f.write(bdata['content']) count += 1 return count def cmd_clone(args): if not args: die('usage: forge clone <owner/repo> [dest]') spec = args[0] # accept a bare "owner/repo", or a full LuisForge clone URL if '/' in spec and spec.startswith(('http://', 'https://')): path = urllib.parse.urlparse(spec).path # /git/<owner>/<repo>.git parts = [p for p in path.split('/') if p] if len(parts) < 3: die(f'could not parse repo from URL: {spec}') owner, name = parts[-2], parts[-1].removesuffix('.git') else: if '/' not in spec: die('expected format: owner/repo') owner, name = spec.split('/', 1) dest = args[1] if len(args) > 1 else name if os.path.exists(dest) and os.listdir(dest): die(f"'{dest}' already exists and isn't empty") os.makedirs(dest, exist_ok=True) status, meta = request('GET', f'/api/repos/{owner}/{name}', auth=False) if status != 200: die(meta.get('error', 'repo not found')) print(f"Cloning {owner}/{name} into '{dest}'...") count = fetch_dir(owner, name, '', dest) print(f"Done — {count} file(s) written.") # ---------------- self-update ---------------- def cmd_update(args): force = '--force' in args this_path = os.path.realpath(__file__) try: req = urllib.request.Request(UPDATE_URL, headers={'User-Agent': USER_AGENT}) with urllib.request.urlopen(req, timeout=10) as resp: remote_bytes = resp.read() except urllib.error.URLError as e: die(f"couldn't reach {UPDATE_URL} ({e.reason})") with open(this_path, 'rb') as f: local_bytes = f.read() if not force and hashlib.sha256(remote_bytes).digest() == hashlib.sha256(local_bytes).digest(): print(f'Already up to date (v{VERSION}).') return tmp_path = this_path + '.new' with open(tmp_path, 'wb') as f: f.write(remote_bytes) os.chmod(tmp_path, 0o755) os.replace(tmp_path, this_path) print(f'Updated {this_path} from {UPDATE_URL}.') print("Run 'forge -h' to confirm it's the version you expect.") # ---------------- uninstall ---------------- PATH_MARKER = '# added by LuisForge install.sh' def _strip_installer_path_line(rc_path): if not os.path.exists(rc_path): return False with open(rc_path) as f: lines = f.readlines() out = [] changed = False skip_next_blank = False i = 0 while i < len(lines): line = lines[i] if line.strip() == PATH_MARKER: changed = True i += 1 # also drop the export line right after the marker, and one # leading blank line the installer adds before the marker if i < len(lines) and 'export PATH=' in lines[i]: i += 1 if out and out[-1].strip() == '': out.pop() continue out.append(line) i += 1 if changed: with open(rc_path, 'w') as f: f.writelines(out) return changed def cmd_uninstall(args): this_path = os.path.realpath(__file__) codey_path = os.path.join(os.path.dirname(this_path), 'codey') has_codey = os.path.exists(codey_path) if '--yes' not in args and '-y' not in args: extra = f' and {codey_path}' if has_codey else '' answer = input(f"Remove {this_path}{extra} and ~/.forge (config + saved token)? [y/N] ").strip().lower() if answer not in ('y', 'yes'): print('Aborted.') return if os.path.isdir(CONFIG_DIR): import shutil shutil.rmtree(CONFIG_DIR) print(f'Removed {CONFIG_DIR}') for rc in (os.path.expanduser('~/.zshrc'), os.path.expanduser('~/.bashrc'), os.path.expanduser('~/.profile')): if _strip_installer_path_line(rc): print(f'Removed PATH entry from {rc}') if has_codey: print(f'Removing {codey_path}') os.remove(codey_path) print(f'Removing {this_path}') os.remove(this_path) print('forge has been uninstalled.') # ---------------- dispatch ---------------- COMMANDS = { ('auth', 'signup'): cmd_auth_signup, ('auth', 'login'): cmd_auth_login, ('auth', 'logout'): cmd_auth_logout, ('auth', 'whoami'): cmd_auth_whoami, } TOP_LEVEL = { 'clone': cmd_clone, 'repos': cmd_repos, 'create': cmd_create, 'delete': cmd_delete, 'push': cmd_push, 'files': cmd_files, 'cat': cmd_cat, 'log': cmd_log, 'issues': cmd_issues, 'run': cmd_run, 'cmd': cmd_cmd, '2fa': cmd_2fa, 'ssh-key': cmd_ssh_key, 'update': cmd_update, 'uninstall': cmd_uninstall, } def main(): argv = sys.argv[1:] if not argv or argv[0] in ('-h', '--help'): print(__doc__) return if argv[0] == 'auth': if len(argv) < 2 or (argv[0], argv[1]) not in COMMANDS: print("usage: forge auth <signup|login|logout|whoami>", file=sys.stderr) sys.exit(1) COMMANDS[(argv[0], argv[1])](argv[2:]) return if argv[0] in TOP_LEVEL: TOP_LEVEL[argv[0]](argv[1:]) return print(f"unknown command: {argv[0]}", file=sys.stderr) print(__doc__) sys.exit(1) if __name__ == '__main__': main()