#!/usr/bin/env python3 """ GHL OAuth callback handler for Rob's marketplace app. Catches GET /ghl/oauth/callback?code=...&state=..., exchanges the code for access+refresh tokens at GHL, stores them at /root/.hermes/secrets/ghl-oauth.json (chmod 600), and refreshes the access token in the background before expiry. Config: /root/.hermes/secrets/ghl-oauth.env GHL_CLIENT_ID=... GHL_CLIENT_SECRET=... GHL_REDIRECT_URI=https://robblake.cloud/ghl/oauth/callback (must match app registration EXACTLY) Runs on 127.0.0.1:9120 behind nginx. systemd unit: ghl-oauth.service """ import json import os import stat import subprocess import threading import time import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ENV_FILE = "/root/.hermes/secrets/ghl-oauth.env" TOKEN_FILE = "/root/.hermes/secrets/ghl-oauth.json" GHL_TOKEN_URL = "https://services.leadconnectorhq.com/oauth/token" GHL_API_BASE = "https://services.leadconnectorhq.com" API_VERSION = "2021-07-28" LISTEN = ("127.0.0.1", 9120) # Refresh when fewer than this many seconds remain before expiry REFRESH_MARGIN = 3600 REFRESH_CHECK_INTERVAL = 600 def load_env(path): cfg = {} with open(path) as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) cfg[k.strip()] = v.strip() return cfg CFG = load_env(ENV_FILE) CLIENT_ID = CFG["GHL_CLIENT_ID"] CLIENT_SECRET = CFG["GHL_CLIENT_SECRET"] REDIRECT_URI = CFG["GHL_REDIRECT_URI"] def log(msg): print(f"[ghl-oauth] {msg}", flush=True) def post_form(url, data): body = urllib.parse.urlencode(data) cmd = [ "curl", "-s", "-X", "POST", url, "-H", "Content-Type: application/x-www-form-urlencoded", "-H", "Accept: application/json", "-d", body, "--max-time", "30", ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=35) if result.returncode != 0: raise RuntimeError(f"curl failed: {result.stderr}") return json.loads(result.stdout) def save_tokens(payload): """Persist token payload atomically with 600 perms, adding absolute expiry.""" payload = dict(payload) payload["obtained_at"] = int(time.time()) if "expires_in" in payload: payload["expires_at"] = payload["obtained_at"] + int(payload["expires_in"]) tmp = TOKEN_FILE + ".tmp" with open(tmp, "w") as f: json.dump(payload, f, indent=2) os.chmod(tmp, stat.S_IRUSR | stat.S_IWUSR) os.replace(tmp, TOKEN_FILE) log(f"tokens saved: locationId={payload.get('locationId')} " f"companyId={payload.get('companyId')} userType={payload.get('userType')} " f"scope={str(payload.get('scope'))[:80]}") def exchange_code(code): return post_form(GHL_TOKEN_URL, { "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URI, }) def refresh_tokens(refresh_token): return post_form(GHL_TOKEN_URL, { "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "grant_type": "refresh_token", "refresh_token": refresh_token, }) def refresh_loop(): """Background thread: keep the access token fresh.""" while True: time.sleep(REFRESH_CHECK_INTERVAL) try: if not os.path.exists(TOKEN_FILE): continue with open(TOKEN_FILE) as f: tok = json.load(f) expires_at = tok.get("expires_at", 0) remaining = expires_at - time.time() if remaining > REFRESH_MARGIN: continue rt = tok.get("refresh_token") if not rt: log("WARNING: token near expiry but no refresh_token stored") continue log(f"access token expires in {int(remaining)}s — refreshing") new = refresh_tokens(rt) save_tokens(new) log("refresh OK") except Exception as e: log(f"refresh error: {e}") class Handler(BaseHTTPRequestHandler): server_version = "ghl-oauth/1.0" def _send(self, code, body, ctype="application/json"): data = body.encode() self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def do_GET(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == "/ghl/health": state = {"status": "ok", "token_file": os.path.exists(TOKEN_FILE)} if state["token_file"]: with open(TOKEN_FILE) as f: tok = json.load(f) state["locationId"] = tok.get("locationId") state["expires_in_s"] = int(tok.get("expires_at", 0) - time.time()) self._send(200, json.dumps(state)) return if parsed.path != "/ghl/oauth/callback": self._send(404, json.dumps({"error": "not found"})) return qs = urllib.parse.parse_qs(parsed.query) err = qs.get("error", [None])[0] if err: log(f"oauth error from GHL: {err} {qs.get('error_description')}") self._send(400, json.dumps({"error": err, "detail": qs.get("error_description", [""])[0]})) return code = qs.get("code", [None])[0] if not code: self._send(400, json.dumps({"error": "missing code parameter"})) return log("received authorization code — exchanging") try: tokens = exchange_code(code) save_tokens(tokens) except urllib.error.HTTPError as e: detail = e.read().decode()[:500] log(f"exchange failed: HTTP {e.code} {detail}") self._send(502, json.dumps({"error": "token exchange failed", "ghl_status": e.code, "detail": detail})) return except Exception as e: log(f"exchange failed: {e}") self._send(502, json.dumps({"error": "token exchange failed", "detail": str(e)})) return self._send(200, """GHL connected

✅ App installed

Your GoHighLevel app is now connected. Tokens are stored on the server and will refresh automatically.

You can close this tab.

""", "text/html") def log_message(self, fmt, *args): log("http " + (fmt % args)) if __name__ == "__main__": threading.Thread(target=refresh_loop, daemon=True).start() log(f"listening on {LISTEN[0]}:{LISTEN[1]}") ThreadingHTTPServer(LISTEN, Handler).serve_forever()