feat(auth): support Machine Identity JWT login

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!

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)

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

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

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

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