Skip to content

Commit b6b67e3

Browse files
jamesfyushrijeetj
authored andcommitted
feat: Add SageMaker token generator to sagemaker-core
Embed the aws-sagemaker-token-generator library into sagemaker.core so users can generate SageMaker bearer tokens without installing a separate wheel. Usage: from sagemaker.core.aws_sagemaker_token_generator import provide_token token = provide_token(region='us-east-1')
1 parent bb0b615 commit b6b67e3

2 files changed

Lines changed: 141 additions & 0 deletions

File tree

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
"""AWS SageMaker Token Generator.
2+
3+
A lightweight module for generating short-term bearer tokens for AWS SageMaker
4+
API authentication. Provides the ``generate_token`` helper and the lower-level
5+
``SageMakerTokenGenerator`` class.
6+
7+
Example::
8+
9+
>>> from sagemaker.core.token_generator import generate_token
10+
>>> token = generate_token(region="us-east-1")
11+
"""
12+
13+
from __future__ import annotations
14+
15+
import os
16+
from datetime import timedelta
17+
18+
from botocore.credentials import CredentialProvider
19+
from botocore.session import Session
20+
21+
from sagemaker.core.token_generator.token_generator import (
22+
TOKEN_DURATION,
23+
SageMakerTokenGenerator,
24+
_generate_token,
25+
)
26+
27+
__all__ = ["SageMakerTokenGenerator", "generate_token"]
28+
29+
30+
def generate_token(
31+
region: str | None = None,
32+
aws_credentials_provider: CredentialProvider | None = None,
33+
expiry: timedelta = timedelta(hours=12),
34+
) -> str:
35+
"""Generate a short-lived AWS SageMaker bearer token.
36+
37+
Args:
38+
region (str): AWS region. Falls back to the ``AWS_REGION``
39+
environment variable when not provided.
40+
aws_credentials_provider (CredentialProvider): Optional credential
41+
provider. Uses the default AWS credential chain when omitted.
42+
expiry (timedelta): Token lifetime. Must be between 1 second and
43+
12 hours inclusive. Defaults to 12 hours.
44+
45+
Returns:
46+
str: A bearer token string.
47+
48+
Raises:
49+
ValueError: If *region* is missing or *expiry* is out of range.
50+
RuntimeError: If no valid AWS credentials are found.
51+
"""
52+
region = region or os.environ.get("AWS_REGION")
53+
if not region:
54+
raise ValueError("Region must be provided or set via the AWS_REGION environment variable.")
55+
56+
if expiry.total_seconds() <= 0 or expiry.total_seconds() > TOKEN_DURATION:
57+
raise ValueError(
58+
"Token expiry must be greater than zero and less than or equal to 12 hours"
59+
)
60+
61+
credentials = (
62+
aws_credentials_provider.load() if aws_credentials_provider else Session().get_credentials()
63+
)
64+
65+
if credentials is None:
66+
raise RuntimeError(
67+
"No AWS credentials found. Check your environment or credential provider."
68+
)
69+
70+
return _generate_token(credentials, region, int(expiry.total_seconds()))
Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
"""SageMaker Token Generator core signing logic.
2+
3+
Generates short-term bearer tokens for AWS SageMaker API authentication
4+
using SigV4 signed pre-signed URLs.
5+
"""
6+
7+
from __future__ import annotations
8+
9+
import base64
10+
11+
from botocore.auth import SigV4QueryAuth
12+
from botocore.awsrequest import AWSRequest
13+
from botocore.credentials import Credentials
14+
15+
DEFAULT_HOST: str = "sagemaker.amazonaws.com"
16+
DEFAULT_URL: str = f"https://{DEFAULT_HOST}/"
17+
SERVICE_NAME: str = "sagemaker"
18+
AUTH_PREFIX: str = "sagemaker-api-key-"
19+
TOKEN_VERSION: str = "&Version=1"
20+
TOKEN_DURATION: int = 43200 # 12 hours in seconds
21+
22+
23+
def _generate_token(credentials: Credentials, region: str, expires: int) -> str:
24+
"""Build a presigned bearer token.
25+
26+
Args:
27+
credentials (Credentials): AWS credentials.
28+
region (str): AWS region.
29+
expires (int): Expiry time in seconds.
30+
31+
Returns:
32+
str: A base64-encoded bearer token string.
33+
"""
34+
request = AWSRequest(
35+
method="POST",
36+
url=DEFAULT_URL,
37+
headers={"host": DEFAULT_HOST},
38+
params={"Action": "CallWithBearerToken"},
39+
)
40+
41+
auth = SigV4QueryAuth(credentials, SERVICE_NAME, region, expires=expires)
42+
auth.add_auth(request)
43+
44+
presigned_url = request.url.replace("https://", "") + TOKEN_VERSION
45+
encoded_token = base64.b64encode(presigned_url.encode("utf-8")).decode("utf-8")
46+
47+
return f"{AUTH_PREFIX}{encoded_token}"
48+
49+
50+
class SageMakerTokenGenerator:
51+
"""Generate short-lived AWS SageMaker bearer tokens."""
52+
53+
def get_token(self, credentials: Credentials, region: str) -> str:
54+
"""Generate a token using provided credentials and region.
55+
56+
Args:
57+
credentials (Credentials): AWS credentials to sign the request.
58+
region (str): AWS region.
59+
60+
Returns:
61+
str: A bearer token string.
62+
63+
Raises:
64+
ValueError: If inputs are invalid.
65+
"""
66+
if not credentials:
67+
raise ValueError("Credentials cannot be None")
68+
if not region or not isinstance(region, str):
69+
raise ValueError("Region must be a non-empty string")
70+
71+
return _generate_token(credentials, region, TOKEN_DURATION)

0 commit comments

Comments
 (0)