#!/usr/bin/env python3 import argparse import datetime as dt import json import os from pathlib import Path import sys import urllib.error import urllib.request DEFAULT_BASE_URL = os.environ.get("CLOCKP_OAUTH_BASE_URL", "https://feishu-oauth.dev.clock-p.com") def resolve_token_file(local: bool) -> Path: if local: return Path.cwd() / ".dev.clock-p.com" / "feishu-token" if os.name == "nt": appdata = os.environ.get("APPDATA") base = Path(appdata) if appdata else (Path.home() / "AppData" / "Roaming") return base / "dev.clock-p.com" / "feishu-token" return Path.home() / ".dev.clock-p.com" / "feishu-token" def ensure_parent(path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) if os.name != "nt": os.chmod(path.parent, 0o700) def read_token(path: Path) -> str: if not path.exists(): return "" return path.read_text(encoding="utf-8").strip() def save_token(path: Path, token: str) -> None: ensure_parent(path) path.write_text(token.strip() + "\n", encoding="utf-8") if os.name != "nt": os.chmod(path, 0o600) def request_json(method: str, url: str, payload=None, bearer_token: str = ""): data = None headers = {"Accept": "application/json"} if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" if bearer_token: headers["Authorization"] = f"Bearer {bearer_token}" req = urllib.request.Request(url, data=data, headers=headers, method=method.upper()) try: with urllib.request.urlopen(req, timeout=30) as resp: body = resp.read().decode("utf-8", errors="replace") if not body.strip(): return {} return json.loads(body) except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") message = f"http {exc.code}" try: parsed = json.loads(body) if body.strip() else {} if isinstance(parsed, dict): message = str(parsed.get("message") or parsed.get("error") or message) except Exception: if body.strip(): message = body.strip() raise RuntimeError(message) from exc except urllib.error.URLError as exc: raise RuntimeError(f"request failed: {exc.reason}") from exc def fetch_me(base_url: str, token: str): if not token: return None url = f"{base_url.rstrip('/')}/api/auth/me" try: data = request_json("GET", url, bearer_token=token) except RuntimeError: return None if not isinstance(data, dict) or not data.get("ok"): return None user = data.get("user") if not isinstance(user, dict): return None return user def looks_like_token(value: str) -> bool: return value.count(".") == 1 and len(value) > 32 def fmt_exp_utc(exp: object) -> str: if not isinstance(exp, int): return "" return dt.datetime.fromtimestamp(exp, tz=dt.timezone.utc).isoformat() def cmd_login(base_url: str, token_file: Path, with_token: bool) -> int: existing = read_token(token_file) if existing: me = fetch_me(base_url, existing) if me: principal = str(me.get("principalId") or me.get("username") or "") print(f"already login: {principal}") print(f"token file: {token_file}") return 0 start_url = f"{base_url.rstrip('/')}/api/nonweb/flow/start" start = request_json("POST", start_url, payload={"showTokenOnPage": with_token}) flow_id = str(start.get("flowId") or "").strip() verify_url = str(start.get("verifyUrl") or "").strip() if not flow_id or not verify_url: raise RuntimeError("invalid flow response") print("open this url in browser and finish Feishu OAuth:") print(verify_url) pasted = input("paste code or token: " if with_token else "paste code: ").strip() if not pasted: raise RuntimeError("empty input") if looks_like_token(pasted): token = pasted else: exchange_url = f"{base_url.rstrip('/')}/api/nonweb/flow/exchange" exchanged = request_json("POST", exchange_url, payload={"flowId": flow_id, "code": pasted}) token = str(exchanged.get("accessToken") or "").strip() if not token: raise RuntimeError("exchange succeeded but token is empty") me = fetch_me(base_url, token) if not me: raise RuntimeError("token verification failed") save_token(token_file, token) principal = str(me.get("principalId") or me.get("username") or "") exp_utc = fmt_exp_utc(me.get("exp")) print(f"login success: {principal}") if exp_utc: print(f"token exp (utc): {exp_utc}") print(f"token file: {token_file}") return 0 def cmd_logout(token_file: Path) -> int: if token_file.exists(): token_file.unlink() print(f"logout success: removed {token_file}") return 0 print("already logout: token file not found") return 0 def cmd_status(base_url: str, token_file: Path) -> int: token = read_token(token_file) if not token: print("not login") print(f"token file: {token_file}") return 1 me = fetch_me(base_url, token) if not me: print("token exists but invalid/expired") print(f"token file: {token_file}") return 1 principal = str(me.get("principalId") or me.get("username") or "") exp_utc = fmt_exp_utc(me.get("exp")) print(f"login: {principal}") if exp_utc: print(f"token exp (utc): {exp_utc}") print(f"token file: {token_file}") return 0 def main() -> int: parser = argparse.ArgumentParser(description="Feishu OAuth token client for dev.clock-p.com") parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="OAuth service base URL") parser.add_argument( "--local", action="store_true", help="use local token path: ./.dev.clock-p.com/feishu-token", ) sub = parser.add_subparsers(dest="cmd", required=True) login_parser = sub.add_parser("login", help="start non-web login flow") login_parser.add_argument( "--with-token", action="store_true", help="ask server to show token on browser page (default only shows code)", ) sub.add_parser("logout", help="remove local token") sub.add_parser("status", help="print local login status") args = parser.parse_args() token_file = resolve_token_file(args.local) if args.cmd == "login": return cmd_login(args.base_url, token_file, bool(getattr(args, "with_token", False))) if args.cmd == "logout": return cmd_logout(token_file) if args.cmd == "status": return cmd_status(args.base_url, token_file) print(f"unsupported command: {args.cmd}", file=sys.stderr) return 2 if __name__ == "__main__": try: raise SystemExit(main()) except RuntimeError as exc: print(f"error: {exc}", file=sys.stderr) raise SystemExit(1)