| 1 | #!/usr/bin/env python3
|
| 2 | """Bearer token for the Superset MCP server (https://superset.swiftdrive-tech.ru/mcp).
|
| 3 |
|
| 4 | Keycloak realm `infra`, public client `superset-mcp`, Device Authorization Grant.
|
| 5 | First run opens a browser login (realm 2FA applies); after that the cached
|
| 6 | refresh token renews silently until the Keycloak session ends.
|
| 7 |
|
| 8 | mcp-token.py print the access token
|
| 9 | mcp-token.py --header print {"Authorization": "Bearer ..."} (Claude Code headersHelper)
|
| 10 | mcp-token.py --logout drop the cached token
|
| 11 | """
|
| 12 | import json
|
| 13 | import os
|
| 14 | import subprocess
|
| 15 | import sys
|
| 16 | import time
|
| 17 | import urllib.error
|
| 18 | import urllib.parse
|
| 19 | import urllib.request
|
| 20 |
|
| 21 | ISSUER = os.environ.get("SUPERSET_MCP_ISSUER", "https://auth.swiftdrive-tech.ru/realms/infra")
|
| 22 | CLIENT_ID = os.environ.get("SUPERSET_MCP_CLIENT_ID", "superset-mcp")
|
| 23 | CACHE = os.path.expanduser("~/.cache/superset-mcp/token.json")
|
| 24 | MIN_TTL = 300 # renew when fewer seconds than this remain
|
| 25 |
|
| 26 |
|
| 27 | def post(path, **form):
|
| 28 | body = urllib.parse.urlencode({"client_id": CLIENT_ID, **form}).encode()
|
| 29 | req = urllib.request.Request(f"{ISSUER}/protocol/openid-connect/{path}", body)
|
| 30 | try:
|
| 31 | with urllib.request.urlopen(req, timeout=30) as resp:
|
| 32 | return json.load(resp)
|
| 33 | except urllib.error.HTTPError as e:
|
| 34 | try:
|
| 35 | return json.load(e)
|
| 36 | except ValueError:
|
| 37 | return {"error": f"HTTP {e.code}"}
|
| 38 | except urllib.error.URLError as e:
|
| 39 | sys.exit(f"superset-mcp: {ISSUER} unreachable: {e.reason}")
|
| 40 |
|
| 41 |
|
| 42 | def save(tok):
|
| 43 | now = time.time()
|
| 44 | tok["expires_at"] = now + tok["expires_in"]
|
| 45 | tok["refresh_expires_at"] = now + tok.get("refresh_expires_in", 0)
|
| 46 | cache_dir = os.path.dirname(CACHE)
|
| 47 | os.makedirs(cache_dir, mode=0o700, exist_ok=True)
|
| 48 | os.chmod(cache_dir, 0o700)
|
| 49 | tmp = f"{CACHE}.{os.getpid()}"
|
| 50 | fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
| 51 | with os.fdopen(fd, "w") as f:
|
| 52 | json.dump(tok, f)
|
| 53 | os.replace(tmp, CACHE) # concurrent headersHelper runs never see a torn file
|
| 54 | return tok
|
| 55 |
|
| 56 |
|
| 57 | def load():
|
| 58 | try:
|
| 59 | with open(CACHE) as f:
|
| 60 | return json.load(f)
|
| 61 | except (OSError, ValueError):
|
| 62 | return None
|
| 63 |
|
| 64 |
|
| 65 | def device_login():
|
| 66 | dev = post("auth/device")
|
| 67 | if "device_code" not in dev:
|
| 68 | sys.exit(f"device authorization failed: {dev}")
|
| 69 | url = dev.get("verification_uri_complete") or dev["verification_uri"]
|
| 70 | print(f"Superset MCP login: open {url} (code {dev['user_code']})", file=sys.stderr)
|
| 71 | if sys.platform == "darwin":
|
| 72 | subprocess.run(["open", url], check=False)
|
| 73 | interval = dev.get("interval", 5)
|
| 74 | deadline = time.time() + dev.get("expires_in", 600)
|
| 75 | while time.time() < deadline:
|
| 76 | time.sleep(interval)
|
| 77 | tok = post("token", grant_type="urn:ietf:params:oauth:grant-type:device_code",
|
| 78 | device_code=dev["device_code"])
|
| 79 | if "access_token" in tok:
|
| 80 | return save(tok)
|
| 81 | err = tok.get("error")
|
| 82 | if err == "slow_down":
|
| 83 | interval += 5
|
| 84 | elif err != "authorization_pending":
|
| 85 | sys.exit(f"login failed: {tok}")
|
| 86 | sys.exit("login timed out")
|
| 87 |
|
| 88 |
|
| 89 | def token():
|
| 90 | tok = load()
|
| 91 | now = time.time()
|
| 92 | if tok and tok["expires_at"] - now > MIN_TTL:
|
| 93 | return tok
|
| 94 | if tok and tok.get("refresh_token") and tok["refresh_expires_at"] > now:
|
| 95 | new = post("token", grant_type="refresh_token", refresh_token=tok["refresh_token"])
|
| 96 | if "access_token" in new:
|
| 97 | return save(new)
|
| 98 | if not sys.stdin.isatty() and not sys.stderr.isatty():
|
| 99 | # headersHelper runs non-interactively: don't hang on a login nobody sees
|
| 100 | sys.exit("superset-mcp: not logged in, run mcp-token.py in a terminal")
|
| 101 | return device_login()
|
| 102 |
|
| 103 |
|
| 104 | def main():
|
| 105 | arg = sys.argv[1] if len(sys.argv) > 1 else ""
|
| 106 | if arg == "--logout":
|
| 107 | if os.path.exists(CACHE):
|
| 108 | os.remove(CACHE)
|
| 109 | return
|
| 110 | access = token()["access_token"]
|
| 111 | print(json.dumps({"Authorization": f"Bearer {access}"}) if arg == "--header" else access)
|
| 112 |
|
| 113 |
|
| 114 | if __name__ == "__main__":
|
| 115 | main()
|
| 116 |
|
