Files
Brancheneinstufung2/connector-superoffice/auth_handler.py
Floke 7b61426c55 feat(connector-superoffice): implement OAuth 2.0 flow and S2S architecture
Completed POC for SuperOffice integration with the following key achievements:
- Switched from RSA/SOAP to OAuth 2.0 (Refresh Token Flow) for better compatibility with SOD environment.
- Implemented robust token refreshing and caching mechanism in .
- Solved 'Wrong Subdomain' issue by enforcing  for tenant .
- Created  for REST API interaction (Search, Create, Update UDFs).
- Added helper scripts: , , .
- Documented usage and configuration in .
- Updated  configuration requirements.

[2ff88f42]
2026-02-09 16:04:16 +00:00

60 lines
2.1 KiB
Python

import os
import logging
import time
import requests
logger = logging.getLogger(__name__)
class AuthHandler:
def __init__(self):
# Load configuration from environment
self.client_id = os.getenv("SO_CLIENT_ID") or os.getenv("SO_SOD")
self.client_secret = os.getenv("SO_CLIENT_SECRET")
self.refresh_token = os.getenv("SO_REFRESH_TOKEN")
self.tenant_id = os.getenv("SO_CONTEXT_IDENTIFIER") # e.g., Cust55774
# OAuth Token Endpoint for SOD
self.token_url = "https://sod.superoffice.com/login/common/oauth/tokens"
self._access_token = None
self._webapi_url = None
self._expiry = 0
def get_ticket(self):
if self._access_token and time.time() < self._expiry:
return self._access_token, self._webapi_url
return self.refresh_access_token()
def refresh_access_token(self):
logger.info(f"Refreshing Access Token for Client ID: {self.client_id[:5]}...")
payload = {
"grant_type": "refresh_token",
"client_id": self.client_id,
"client_secret": self.client_secret,
"refresh_token": self.refresh_token
}
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Accept": "application/json"
}
try:
resp = requests.post(self.token_url, data=payload, headers=headers, timeout=30)
resp.raise_for_status()
data = resp.json()
self._access_token = data.get("access_token")
# Based on user's browser URL
self._webapi_url = f"https://app-sod.superoffice.com/{self.tenant_id}"
self._expiry = time.time() + int(data.get("expires_in", 3600)) - 60
logger.info("Successfully refreshed Access Token.")
return self._access_token, self._webapi_url
except Exception as e:
logger.error(f"Error refreshing access token: {e}")
if hasattr(e, 'response') and e.response is not None:
logger.error(f"Response: {e.response.text}")
raise