feat: add test_connection command to all integrations for the instance Test button
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,32 @@
|
||||
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/")
|
||||
|
||||
|
||||
def token():
|
||||
data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode()
|
||||
req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST")
|
||||
with urllib.request.urlopen(req, timeout=30) as r:
|
||||
return json.loads(r.read()).get("access_token", "")
|
||||
|
||||
|
||||
def main():
|
||||
# Acquiring the OAuth2 token validates client_id/client_secret: bad
|
||||
# credentials make the exchange fail (HTTPError -> exit 1).
|
||||
tok = token()
|
||||
if not tok:
|
||||
print(json.dumps({"ok": False, "error": "No access token returned"}))
|
||||
sys.exit(1)
|
||||
print(json.dumps({"ok": True, "authenticated": True}))
|
||||
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user