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
6 changes: 4 additions & 2 deletions src/auth0_server_python/auth_server/passwordless_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,15 +85,16 @@ async def start(
MissingRequiredArgumentError: When a magic link is requested but no
``redirect_uri`` is configured on the client, or ``store_options``
is not provided.
ConfigurationError: When client authentication is not configured.
"""
client = self._client
origin_domain = await client._resolve_current_domain(store_options)

body: dict[str, Any] = {
"client_id": client._client_id,
"client_secret": client._client_secret,
"connection": options.connection,
}
client._apply_client_authentication(body, f"https://{origin_domain}/", in_body=True)
Comment thread
kishore7snehil marked this conversation as resolved.
Comment thread
kishore7snehil marked this conversation as resolved.

if isinstance(options, StartPasswordlessEmailOptions):
body["email"] = options.email
Expand Down Expand Up @@ -194,6 +195,7 @@ async def verify(
ApiError: When fetching the JWKS used to verify the ID token fails.
SessionExpiredError: When the token's session-expiry ceiling is
already in the past.
ConfigurationError: When client authentication is not configured.
"""
client = self._client
origin_domain = await client._resolve_current_domain(store_options)
Expand Down Expand Up @@ -222,12 +224,12 @@ async def verify(
body: dict[str, Any] = {
"grant_type": PASSWORDLESS_OTP_GRANT_TYPE,
"client_id": client._client_id,
"client_secret": client._client_secret,
"realm": options.connection,
"username": options.username,
"otp": options.verification_code,
"scope": scope,
}
client._apply_client_authentication(body, origin_issuer or f"https://{origin_domain}/", in_body=True)
if options.audience:
body["audience"] = options.audience

Expand Down
25 changes: 6 additions & 19 deletions src/auth0_server_python/error/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -364,35 +364,22 @@ def __init__(self):
# =============================================================================


class PasswordlessError(ApiError):
"""
Base class for passwordless (embedded login) errors.

Carries the Auth0 ``error`` / ``error_description`` from the API response
body so integrators can branch on a typed exception rather than parsing
strings.
"""

def __init__(self, code: str, message: str, cause=None, retry_after: Optional[int] = None):
super().__init__(code, message, cause)
self.name = "PasswordlessError"
self.retry_after = retry_after


class PasswordlessStartError(PasswordlessError):
class PasswordlessStartError(ApiError):
"""Error raised when POST /passwordless/start fails."""

def __init__(self, code: str, message: str, cause=None, retry_after: Optional[int] = None):
super().__init__(code, message, cause, retry_after)
super().__init__(code, message, cause)
self.name = "PasswordlessStartError"
self.retry_after = retry_after # seconds


class PasswordlessVerifyError(PasswordlessError):
class PasswordlessVerifyError(ApiError):
"""Error raised when the passwordless OTP token exchange fails."""

def __init__(self, code: str, message: str, cause=None, retry_after: Optional[int] = None):
super().__init__(code, message, cause, retry_after)
super().__init__(code, message, cause)
self.name = "PasswordlessVerifyError"
self.retry_after = retry_after # seconds


class PasswordlessErrorCode:
Expand Down
148 changes: 148 additions & 0 deletions src/auth0_server_python/tests/test_passwordless_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@

import jwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from pydantic import ValidationError

from auth0_server_python.auth_server.passwordless_client import (
Expand All @@ -20,6 +22,7 @@
VerifyPasswordlessOtpOptions,
)
from auth0_server_python.error import (
ConfigurationError,
InvalidArgumentError,
IssuerValidationError,
MfaRequiredError,
Expand Down Expand Up @@ -1034,3 +1037,148 @@ async def test_verify_mfa_required_without_token_falls_through(self, mocker):
)
assert exc.value.code == "mfa_required"
client._state_store.set.assert_not_awaited()


# ── Private Key JWT (client assertion) client authentication ────────────────


def _generate_rsa_private_key_pem() -> str:
Comment thread
kishore7snehil marked this conversation as resolved.
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
return key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
).decode("ascii")


def _public_key_pem(private_key_pem: str) -> str:
private_key = serialization.load_pem_private_key(
private_key_pem.encode("ascii"), password=None
)
return private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
).decode("ascii")


class TestPrivateKeyJwt:
@pytest.mark.asyncio
async def test_start_uses_private_key_jwt_assertion(self):
private_key = _generate_rsa_private_key_pem()
client = _make_client(client_secret=None, client_assertion_signing_key=private_key)
http = _mock_http(client, 200, {})

await client.passwordless.start(
StartPasswordlessEmailOptions(email="user@example.com", send="code")
)

body = http.post.call_args.kwargs["json"]
assert "client_secret" not in body
assert body["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
assert len(body["client_assertion"].split(".")) == 3
claims = jwt.decode(
body["client_assertion"],
_public_key_pem(private_key),
algorithms=["RS256"],
audience=f"https://{DOMAIN}/",
)
assert claims["iss"] == CLIENT_ID
assert claims["sub"] == CLIENT_ID

@pytest.mark.asyncio
async def test_verify_uses_private_key_jwt_assertion(self, mocker):
private_key = _generate_rsa_private_key_pem()
client = _make_client(client_secret=None, client_assertion_signing_key=private_key)
claims_from_id_token = {"iss": ISSUER, "sub": "auth0|1", "sid": "SID-123", "iat": 1_000}
mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=METADATA)
mocker.patch.object(
client,
"_get_jwks_cached",
return_value={"keys": [{"kty": "RSA", "kid": "k1"}]},
)
mocker.patch.object(client, "_verify_and_decode_jwt", return_value=claims_from_id_token)
http = _mock_http(
client,
200,
{"access_token": "at", "id_token": "idt", "expires_in": 3600, "scope": "openid"},
)

await client.passwordless.verify(
VerifyPasswordlessOtpOptions(
connection="email", email="user@example.com", verification_code="123456"
)
)

data = http.post.call_args.kwargs["data"]
assert "client_secret" not in data
assert data["client_assertion_type"] == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
assert len(data["client_assertion"].split(".")) == 3
claims = jwt.decode(
data["client_assertion"],
_public_key_pem(private_key),
algorithms=["RS256"],
audience=ISSUER,
)
assert claims["iss"] == CLIENT_ID
assert claims["sub"] == CLIENT_ID

@pytest.mark.asyncio
async def test_start_uses_client_secret_when_configured(self):
client = _make_client()
http = _mock_http(client, 200, {})

await client.passwordless.start(
StartPasswordlessEmailOptions(email="user@example.com", send="code")
)

body = http.post.call_args.kwargs["json"]
assert body["client_secret"] == CLIENT_SECRET
assert "client_assertion" not in body

@pytest.mark.asyncio
async def test_verify_uses_client_secret_when_configured(self, mocker):
client = _make_client()
claims_from_id_token = {"iss": ISSUER, "sub": "auth0|1", "sid": "SID-123", "iat": 1_000}
mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=METADATA)
mocker.patch.object(
client,
"_get_jwks_cached",
return_value={"keys": [{"kty": "RSA", "kid": "k1"}]},
)
mocker.patch.object(client, "_verify_and_decode_jwt", return_value=claims_from_id_token)
http = _mock_http(
client,
200,
{"access_token": "at", "id_token": "idt", "expires_in": 3600, "scope": "openid"},
)

await client.passwordless.verify(
VerifyPasswordlessOtpOptions(
connection="email", email="user@example.com", verification_code="123456"
)
)

data = http.post.call_args.kwargs["data"]
assert data["client_secret"] == CLIENT_SECRET
assert "client_assertion" not in data

@pytest.mark.asyncio
async def test_start_no_client_auth_configured_raises_configuration_error(self):
Comment thread
kishore7snehil marked this conversation as resolved.
client = _make_client(client_secret=None)

with pytest.raises(ConfigurationError):
await client.passwordless.start(
StartPasswordlessEmailOptions(email="user@example.com", send="code")
)

@pytest.mark.asyncio
async def test_verify_no_client_auth_configured_raises_configuration_error(self, mocker):
client = _make_client(client_secret=None)
mocker.patch.object(client, "_get_oidc_metadata_cached", return_value=METADATA)

with pytest.raises(ConfigurationError):
await client.passwordless.verify(
VerifyPasswordlessOtpOptions(
connection="email", email="user@example.com", verification_code="123456"
)
)
Loading