Guillaume BOURGEOIS
84be770928
Gmail for one mailbox over OAuth 2.0 (no service account / delegation): auth-link + exchange-code to obtain a refresh token, connectivity test, message search/get, send/reply with attachments, attachment retrieval, and get_incidents ingestion with an OCSF mapper. Refresh-token grant, stdlib-only (no extra Python dependencies). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
|
|
|
TOKEN_URL = "https://oauth2.googleapis.com/token"
|
|
|
|
|
|
def _cfg():
|
|
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
|
|
|
|
def main():
|
|
cfg = _cfg()
|
|
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
code = str(inputs.get("code") or "")
|
|
if not code:
|
|
raise Exception("code is required")
|
|
if code.lower().startswith("code="):
|
|
code = urllib.parse.unquote(code[len("code="):])
|
|
data = urllib.parse.urlencode({
|
|
"code": code,
|
|
"client_id": str(cfg.get("client_id") or ""),
|
|
"client_secret": str(cfg.get("client_secret") or ""),
|
|
"redirect_uri": str(inputs.get("redirect_uri") or cfg.get("redirect_uri") or "http://localhost"),
|
|
"grant_type": "authorization_code",
|
|
}).encode("utf-8")
|
|
req = urllib.request.Request(TOKEN_URL, data=data,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"},
|
|
method="POST")
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
tok = json.loads(r.read())
|
|
if not tok.get("refresh_token"):
|
|
raise Exception("No refresh_token in response (the code may be used/expired, or the client lacks offline access): " + json.dumps(tok))
|
|
print(json.dumps({
|
|
"refresh_token": tok["refresh_token"],
|
|
"note": "Paste this refresh_token into the instance configuration and save.",
|
|
}))
|
|
|
|
|
|
try:
|
|
main()
|
|
except urllib.error.HTTPError as e:
|
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(json.dumps({"error": str(e)}))
|
|
sys.exit(1)
|