Something went wrong. Try again.
A CLI-only AppView with curses, because... reasons
Something went wrong. Try again.
7.2 kB · 217 lines · Created 11mo ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217#!/usr/bin/env python3"""ATProto CLI AppView with OAuth via local HTTP callbackLogs maximally to ~/.atproto_cli/debug.logRequires: pip install atproto requests"""
import cursesimport sysimport osimport jsonimport tempfileimport subprocessimport webbrowserimport unicodedataimport reimport asynciofrom datetime import datetimefrom pathlib import Pathfrom typing import Optional, Dictfrom http.server import HTTPServer, BaseHTTPRequestHandlerfrom urllib.parse import urlparse, parse_qs
try: from atproto import Client, client_utils from atproto_oauth import OAuthClient from atproto_oauth.stores import MemoryStateStore, MemorySessionStoreexcept ImportError: print("Error: atproto or atproto_oauth library not found. Install with: pip install atproto atproto_oauth") sys.exit(1)
CONFIG_DIR = Path.home() / ".atproto_cli"CONFIG_DIR.mkdir(exist_ok=True)SESSION_FILE = CONFIG_DIR / "session.json"DEBUG_LOG = CONFIG_DIR / "debug.log"
def log_debug(msg: str): with open(DEBUG_LOG, 'a') as f: f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n")
def load_session() -> Optional[Dict]: log_debug(f"Attempting to load session from {SESSION_FILE}") if SESSION_FILE.exists(): try: with open(SESSION_FILE) as f: session = json.load(f) log_debug("Session loaded successfully") return session except Exception as e: log_debug(f"Failed to load session: {type(e).__name__}: {e}") else: log_debug("No session.json found") return None
def save_session(session_string: str): log_debug(f"Saving session to {SESSION_FILE}") with open(SESSION_FILE, 'w') as f: f.write(session_string) log_debug("Session saved successfully")
# ========== Minimal HTTP Server for OAuth ==========class OAuthCallbackHandler(BaseHTTPRequestHandler): """Catches the OAuth redirect with code and state"""
def do_GET(self): parsed = urlparse(self.path) qs = parse_qs(parsed.query) self.server.oauth_params = qs self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(b"<html><body><h1>Login complete!</h1>You may close this tab.</body></html>")
class OAuthHTTPServer(HTTPServer): def __init__(self, server_address, RequestHandlerClass): super().__init__(server_address, RequestHandlerClass) self.oauth_params = None
async def oauth_login(handle: str) -> Client: """Perform OAuth flow via local HTTP server""" REDIRECT_URI = "http://127.0.0.1:5000/callback" SCOPE = "atproto"
# Use localhost as redirect_uri per RFC8252 CLIENT_ID = f"http://127.0.0.1:5000?redirect_uri={REDIRECT_URI}&scope={SCOPE}"
oauth_client = OAuthClient( client_id=CLIENT_ID, redirect_uri=REDIRECT_URI, scope=SCOPE, state_store=MemoryStateStore(), session_store=MemorySessionStore() )
# Start authorization auth_url, state = await oauth_client.start_authorization(handle) log_debug(f"Opening browser for handle {handle}: {auth_url}") webbrowser.open(auth_url)
# Start minimal HTTP server to catch the callback server = OAuthHTTPServer(('127.0.0.1', 5000), OAuthCallbackHandler) log_debug("Waiting for OAuth callback on 127.0.0.1:5000") while server.oauth_params is None: server.handle_request()
params = server.oauth_params code = params.get('code', [None])[0] callback_state = params.get('state', [None])[0] iss = params.get('iss', [None])[0]
if not code or not callback_state or not iss: raise ValueError("OAuth callback missing required parameters")
# Complete OAuth flow oauth_session = await oauth_client.handle_callback(code, callback_state, iss) session_string = oauth_session.export_session_string() save_session(session_string)
log_debug(f"OAuth login complete for handle {oauth_session.handle}") client = Client() client.login(session_string=session_string) return client
# ==================== CLI App Skeleton ====================class ATProtoApp: def __init__(self, stdscr): self.stdscr = stdscr self.client: Optional[Client] = None
curses.curs_set(0) self.stdscr.clear() self.stdscr.refresh() curses.init_pair(1, curses.COLOR_CYAN, curses.COLOR_BLACK)
log_debug("ATProtoApp initialized")
def login(self): session = load_session() if session: try: self.client = Client() self.client.login(session_string=session) log_debug("Session restored successfully") return True except Exception as e: log_debug(f"Session restore failed: {type(e).__name__}: {e}")
# No session -> choose login method self.stdscr.addstr(0, 0, "No session found. Choose login method:\n1. OAuth (Bluesky handle)\n2. Password / App password\nLogin method [default=2]: ") self.stdscr.refresh() curses.echo() choice = self.stdscr.getstr().decode().strip() curses.noecho()
if choice == "1": self.stdscr.addstr("\nEnter your Bluesky handle: ") self.stdscr.refresh() curses.echo() raw_handle = self.stdscr.getstr().decode(errors="ignore").strip() curses.noecho()
# Remove leading @ handle = raw_handle.lstrip('@')
# Normalize Unicode (NFKC) to standard form handle = unicodedata.normalize("NFKC", handle)
# Remove invisible/control characters handle = "".join(ch for ch in handle if not unicodedata.category(ch).startswith("C"))
# Optional: remove any remaining spaces just in case handle = handle.replace(" ", "")
self.stdscr.addstr(f"\nUsing handle: '{handle}'\n") self.stdscr.refresh()
# Run OAuth login self.client = asyncio.run(oauth_login(handle)) return True else: # Password flow self.stdscr.addstr("\nHandle: ") self.stdscr.refresh() curses.echo() handle = self.stdscr.getstr().decode().strip().removeprefix('@') self.stdscr.addstr("\nApp password: ") self.stdscr.refresh() curses.noecho() password = self.stdscr.getstr().decode().strip() self.client = Client() self.client.login(handle, password) session_string = self.client.export_session_string() save_session(session_string) return True
def run(self): if not self.login(): return self.stdscr.addstr("\nLogin successful! Press any key to exit.\n") self.stdscr.getch()
def main(): try: curses.wrapper(lambda stdscr: ATProtoApp(stdscr).run()) except KeyboardInterrupt: log_debug("KeyboardInterrupt caught, exiting") except Exception as e: log_debug(f"Unhandled exception: {type(e).__name__}: {e}")
if __name__ == "__main__": main()