diff --git a/README.md b/README.md index fdefbd4..80e9044 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,23 @@ export REQUESTS_CA_BUNDLE=/etc/pki/tls/certs/ca-bundle.crt These paths are the operating system bundles, so the CA must be added there first (for example with `update-ca-certificates` or `update-ca-trust`). You can also point `REQUESTS_CA_BUNDLE` to a PEM file that holds only your CA. +## JWT authentication + +For a Machine Identity configured with JWT Auth, use `jwt_auth` (OIDC Auth is a separate method): + +```python +from infisical_sdk import InfisicalSDKClient + +with InfisicalSDKClient(host="https://us.infisical.com") as client: + auth = client.auth.jwt_auth.login(identity_id=identity_id, jwt=identity_jwt) + # Subsequent SDK requests use the returned access token automatically. +``` + +Pass `organization_slug="your-org"` to scope the login to an organization. +See the [JWT Auth API reference](https://infisical.com/docs/api-reference/endpoints/jwt-auth/login). + +Run the offline regression tests with `pytest tests` after installing `test-requirements.txt`. + ## Security Please do not file GitHub issues or post on our public forum for security vulnerabilities, as they are public! diff --git a/infisical_sdk/resources/auth.py b/infisical_sdk/resources/auth.py index 30f36b5..0c0b9dc 100644 --- a/infisical_sdk/resources/auth.py +++ b/infisical_sdk/resources/auth.py @@ -4,6 +4,7 @@ from infisical_sdk.resources.auth_methods import OidcAuth from infisical_sdk.resources.auth_methods import TokenAuth from infisical_sdk.resources.auth_methods import LdapAuth +from infisical_sdk.resources.auth_methods import JwtAuth from typing import Callable class Auth: @@ -12,5 +13,6 @@ def __init__(self, requests: InfisicalRequests, setToken: Callable[[str], None]) self.aws_auth = AWSAuth(requests, setToken) self.universal_auth = UniversalAuth(requests, setToken) self.oidc_auth = OidcAuth(requests, setToken) + self.jwt_auth = JwtAuth(requests, setToken) self.token_auth = TokenAuth(setToken) self.ldap_auth = LdapAuth(requests, setToken) \ No newline at end of file diff --git a/infisical_sdk/resources/auth_methods/__init__.py b/infisical_sdk/resources/auth_methods/__init__.py index 02aeef9..79967ea 100644 --- a/infisical_sdk/resources/auth_methods/__init__.py +++ b/infisical_sdk/resources/auth_methods/__init__.py @@ -3,3 +3,4 @@ from .oidc_auth import OidcAuth from .token_auth import TokenAuth from .ldap_auth import LdapAuth +from .jwt_auth import JwtAuth diff --git a/infisical_sdk/resources/auth_methods/jwt_auth.py b/infisical_sdk/resources/auth_methods/jwt_auth.py new file mode 100644 index 0000000..7932314 --- /dev/null +++ b/infisical_sdk/resources/auth_methods/jwt_auth.py @@ -0,0 +1,39 @@ +from typing import Callable, Optional + +from infisical_sdk.api_types import MachineIdentityLoginResponse +from infisical_sdk.infisical_requests import InfisicalRequests + + +class JwtAuth: + def __init__(self, requests: InfisicalRequests, setToken: Callable[[str], None]): + self.requests = requests + self.setToken = setToken + + def login( + self, + identity_id: str, + jwt: str, + organization_slug: Optional[str] = None, + ) -> MachineIdentityLoginResponse: + """Log in a Machine Identity configured with JWT Auth. + + Args: + identity_id: Your Machine Identity ID. + jwt: The JWT accepted by the identity's JWT Auth configuration. + organization_slug: Optional organization slug for the login request. + + Returns: + The access token and related information. The client's token is + updated only after a successful login. + """ + result = self.requests.post( + path="/api/v1/auth/jwt-auth/login", + json={ + "identityId": identity_id, + "jwt": jwt, + "organizationSlug": organization_slug, + }, + model=MachineIdentityLoginResponse, + ) + self.setToken(result.data.accessToken) + return result.data diff --git a/test-requirements.txt b/test-requirements.txt index 3a0d0b9..87f38bc 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,3 +1,4 @@ pytest~=7.1.3 pytest-cov>=2.8.1 pytest-randomly>=3.12.0 +responses>=0.23,<1 diff --git a/tests/test_jwt_auth.py b/tests/test_jwt_auth.py new file mode 100644 index 0000000..bb2d05f --- /dev/null +++ b/tests/test_jwt_auth.py @@ -0,0 +1,67 @@ +import json + +import pytest +import responses + +from infisical_sdk import InfisicalSDKClient +from infisical_sdk.api_types import MachineIdentityLoginResponse +from infisical_sdk.infisical_requests import APIError + +HOST = "https://example.com" +LOGIN = { + "accessToken": "new-access-token", + "expiresIn": 7200, + "accessTokenMaxTTL": 86400, + "tokenType": "Bearer", +} + + +@pytest.mark.parametrize("initial_token", [None, "existing-token"]) +@pytest.mark.parametrize("organization_slug", [None, "example-org"]) +@responses.activate +def test_jwt_login_sets_token_for_later_requests(initial_token, organization_slug): + responses.post(HOST + "/api/v1/auth/jwt-auth/login", json=LOGIN) + responses.get(HOST + "/probe", json={"ok": True}) + with InfisicalSDKClient(host=HOST, token=initial_token, cache_ttl=None) as client: + options = {} if organization_slug is None else {"organization_slug": organization_slug} + result = client.auth.jwt_auth.login( + identity_id="identity-example", jwt="test.jwt.value", **options + ) + assert isinstance(result, MachineIdentityLoginResponse) + assert result.to_dict() == LOGIN + assert client.get_token() == LOGIN["accessToken"] + client.api.get(path="/probe", model=dict) + expected = {"identityId": "identity-example", "jwt": "test.jwt.value"} + if organization_slug is not None: + expected["organizationSlug"] = organization_slug + assert json.loads(responses.calls[0].request.body) == expected + assert responses.calls[1].request.headers["Authorization"] == "Bearer new-access-token" + + +@pytest.mark.parametrize("initial_token", [None, "existing-token"]) +@pytest.mark.parametrize("status", [401, 403]) +@responses.activate +def test_jwt_login_failure_preserves_current_token(initial_token, status): + responses.post( + HOST + "/api/v1/auth/jwt-auth/login", + json={"message": "Login rejected"}, status=status, + ) + with InfisicalSDKClient(host=HOST, token=initial_token, cache_ttl=None) as client: + with pytest.raises(APIError) as error: + client.auth.jwt_auth.login(identity_id="identity-example", jwt="test.jwt.value") + assert error.value.status_code == status + assert client.get_token() == initial_token + expected_header = "Bearer existing-token" if initial_token else None + assert client.api.session.headers.get("Authorization") == expected_header + assert len(responses.calls) == 1 + + +@pytest.mark.parametrize("initial_token", [None, "existing-token"]) +@responses.activate +def test_oidc_login_keeps_its_own_endpoint(initial_token): + responses.post(HOST + "/api/v1/auth/oidc-auth/login", json=LOGIN) + with InfisicalSDKClient(host=HOST, token=initial_token, cache_ttl=None) as client: + result = client.auth.oidc_auth.login(identity_id="identity-example", jwt="test.jwt.value") + assert result.accessToken == LOGIN["accessToken"] + assert client.get_token() == LOGIN["accessToken"] + assert len(responses.calls) == 1