Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions centml/sdk/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,9 @@ def get_deployment_usage(
step=step,
).values

def initialize_user(self):
return self._api.setup_stripe_customer_payments_setup_post()


@contextmanager
def get_centml_client():
Expand Down
42 changes: 40 additions & 2 deletions centml/sdk/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,19 +58,57 @@ def load_centml_cred():


def get_centml_token():
# Always use fresh client credentials if available
if settings.CENTML_SERVICE_ACCOUNT_ID and settings.CENTML_SERVICE_ACCOUNT_SECRET:
access_token = authenticate_with_client_credentials()
if access_token is not None:
return access_token
else:
sys.exit(
"Could not authenticate with client credentials. Please check your service account configuration..."
)

# Fall back to stored credentials for interactive flows
cred = load_centml_cred()
if not cred:
sys.exit("CentML credentials not found. Please login...")
exp_time = int(jwt.decode(cred["access_token"], options={"verify_signature": False})["exp"])

if time.time() >= exp_time - 100:
cred = refresh_centml_token(cred["refresh_token"])
if cred is None:
# Check if we have a refresh token (interactive flow)
refresh_token = cred.get("refresh_token")
if refresh_token is not None:
# Use refresh token for interactive authentication
cred = refresh_centml_token(cred["refresh_token"])
if cred is None:
sys.exit("Could not refresh credentials. Please login and try again...")
else:
sys.exit("Could not refresh credentials. Please login and try again...")

return cred["access_token"]


def authenticate_with_client_credentials():
"""
Authenticate using client credentials flow for service-to-service authentication.
Returns access token if successful, None otherwise.
"""
if not settings.CENTML_SERVICE_ACCOUNT_ID or not settings.CENTML_SERVICE_ACCOUNT_SECRET:
return None

params = {
'grant_type': 'client_credentials',
'client_id': settings.CENTML_SERVICE_ACCOUNT_ID,
'client_secret': settings.CENTML_SERVICE_ACCOUNT_SECRET,
'scope': 'openid profile email',
}
response = requests.post(settings.CENTML_SERVICE_ACCOUNT_TOKEN_URL, data=params, timeout=10)
response.raise_for_status()
response_data = response.json()
access_token = response_data.get('access_token')
return access_token


def remove_centml_cred():
if os.path.exists(settings.CENTML_CRED_FILE_PATH):
os.remove(settings.CENTML_CRED_FILE_PATH)
8 changes: 8 additions & 0 deletions centml/sdk/config.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import os
from typing import Optional
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict

Expand All @@ -16,5 +17,12 @@ class Config(BaseSettings):

CENTML_WORKOS_CLIENT_ID: str = os.getenv("CENTML_WORKOS_CLIENT_ID", default="client_01JP5TWW2997MF8AYQXHJEGYR0")

# Long-term credentials - can be set via environment variables
CENTML_SERVICE_ACCOUNT_SECRET: Optional[str] = os.getenv("CENTML_SERVICE_ACCOUNT_SECRET", default=None)
CENTML_SERVICE_ACCOUNT_ID: Optional[str] = os.getenv("CENTML_SERVICE_ACCOUNT_ID", default=None)
CENTML_SERVICE_ACCOUNT_TOKEN_URL: str = os.getenv(
"CENTML_SERVICE_ACCOUNT_TOKEN_URL", default="https://signin.centml.com/oauth2/token"
)


settings = Config()
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@

setup(
name='centml',
version='0.4.4',
version='0.4.5',
packages=find_packages(),
python_requires=">=3.10",
long_description=open('README.md').read(),
Expand Down
Loading