From 7d3208d00437e0f9bd2aa6fabdadded97c5ad7b9 Mon Sep 17 00:00:00 2001 From: Scott Wilson Date: Fri, 30 Oct 2026 16:21:34 -0700 Subject: [PATCH] feat(agent-vault): create sessions on behalf of users Co-Authored-By: Claude Opus 5.5 --- infisical_sdk/api_types.py | 7 ++-- infisical_sdk/resources/agent_vault.py | 32 +++++++++++++++---- tests/test_agent_vault.py | 44 ++++++++++++++++++++++++++ 3 files changed, 75 insertions(+), 8 deletions(-) diff --git a/infisical_sdk/api_types.py b/infisical_sdk/api_types.py index f6896c2..2444e82 100644 --- a/infisical_sdk/api_types.py +++ b/infisical_sdk/api_types.py @@ -518,7 +518,7 @@ def from_dict(cls, data: Dict) -> 'ListAgentVaultProxiesResponse': @dataclass class AgentVaultSessionActor(BaseModel): - """The user or machine identity an Agent Vault session belongs to""" + """The user or machine identity an Agent Vault session belongs to, or the user it was started on behalf of""" type: str id: Optional[str] = None username: Optional[str] = None @@ -545,13 +545,16 @@ class AgentVaultSession(BaseModel): expiresAt: Optional[str] = None revokedAt: Optional[str] = None accessBundles: Optional[List[AgentVaultSessionAccessBundle]] = None + onBehalfOf: Optional[AgentVaultSessionActor] = None @classmethod def from_dict(cls, data: Dict) -> 'AgentVaultSession': - """Create model from dictionary with nested actor and access bundles. accessBundles is None if the response leaves it out.""" + """Create model from dictionary with nested actor, on-behalf-of user and access bundles. accessBundles is None if the response leaves it out.""" session_data = data.copy() if data.get('actor') is not None: session_data['actor'] = AgentVaultSessionActor.from_dict(data['actor']) + if data.get('onBehalfOf') is not None: + session_data['onBehalfOf'] = AgentVaultSessionActor.from_dict(data['onBehalfOf']) if data.get('accessBundles') is not None: session_data['accessBundles'] = [ AgentVaultSessionAccessBundle.from_dict(bundle) for bundle in data['accessBundles'] diff --git a/infisical_sdk/resources/agent_vault.py b/infisical_sdk/resources/agent_vault.py index f0938d3..2738f4d 100644 --- a/infisical_sdk/resources/agent_vault.py +++ b/infisical_sdk/resources/agent_vault.py @@ -209,7 +209,7 @@ def list( Args: scope: Whose sessions to list: your own ("mine") or everyone's ("all", administrators only). Defaults to "mine". statuses: Only return sessions with any of these statuses: "active", "revoked" or "expired". - search: Match sessions by label, actor name, actor email or access bundle name. + search: Match sessions by label, actor name, actor email, the name or email of the user a session was started on behalf of, or access bundle name. limit: The maximum number of sessions to return (1-100). Defaults to 20. offset: How many sessions to skip. Defaults to 0. @@ -255,22 +255,42 @@ def get(self, session_id: str) -> AgentVaultSession: def create( self, access_bundle: str, - ttl: str = "7d", - label: Optional[str] = None) -> CreatedAgentVaultSession: - """Create an Agent Vault session with an access bundle you can reach. + ttl: Optional[str] = None, + label: Optional[str] = None, + on_behalf_of_user_email: Optional[str] = None, + on_behalf_of_user_id: Optional[str] = None) -> CreatedAgentVaultSession: + """Create an Agent Vault session with an access bundle you can reach, or, as a machine identity with a delegation, on behalf of a user. + + To start a session on behalf of a user, you must be a machine identity whose delegation covers the user. The proxy + then uses the user's personal values, and the session can use only an access bundle that both the user and your + machine identity can use. The session still belongs to your machine identity, so the user can't see it. Infisical + versions that don't support sessions on behalf of users ignore on_behalf_of_user_email and on_behalf_of_user_id, + and create the session for your machine identity instead. Args: access_bundle: The name of the access bundle the session carries. - ttl: How long the session lasts: a duration such as "30m", "8h" or "7d" (at least 1m), or "never". Defaults to "7d". + ttl: How long the session lasts: a duration such as "30m", "8h" or "7d" (at least 1m), or "never". Defaults to "7d", or to "1h" for a session on behalf of a user. label: A short description of what the session is for, such as "nightly deploy". Up to 128 characters, without line breaks, tabs, or invisible characters. + on_behalf_of_user_email: The email address of the user to start the session on behalf of. Can't be combined with on_behalf_of_user_id. + on_behalf_of_user_id: The ID of the user to start the session on behalf of. Can't be combined with on_behalf_of_user_email. Returns: The created session, including its token. The token is only returned once. """ + if on_behalf_of_user_email is not None and on_behalf_of_user_id is not None: + raise ValueError("Pass either on_behalf_of_user_email or on_behalf_of_user_id, not both") + + on_behalf_of = None + if on_behalf_of_user_email is not None: + on_behalf_of = {"userEmail": on_behalf_of_user_email} + elif on_behalf_of_user_id is not None: + on_behalf_of = {"userId": on_behalf_of_user_id} + request_body = { "accessBundles": [access_bundle], "ttl": ttl, "label": label, + "onBehalfOf": on_behalf_of, } result = self.requests.post_once( @@ -286,7 +306,7 @@ def update(self, session_id: str, label: Optional[str]) -> AgentVaultSession: Args: session_id: The ID of the session to update. - label: The new label for the session, up to 128 characters, without line breaks, tabs, or invisible characters. Pass None or an empty string to remove the label. You can't change the label once the session's owner is deleted. + label: The new label for the session, up to 128 characters, without line breaks, tabs, or invisible characters. Pass None or an empty string to remove the label. You can't change the label once the session's owner, or the user it was started on behalf of, is deleted. Returns: The updated session. diff --git a/tests/test_agent_vault.py b/tests/test_agent_vault.py index 7adad32..0dcb555 100644 --- a/tests/test_agent_vault.py +++ b/tests/test_agent_vault.py @@ -1,6 +1,7 @@ import pytest import requests +from infisical_sdk.api_types import AgentVaultSessionActor from infisical_sdk.resources.agent_vault import AgentVault PROXY = {"id": "a2b5e0d1-6f3e-4c55-9a43-3f0c1f0a2b7e", "name": "edge-proxy"} @@ -11,6 +12,14 @@ "createdAt": "2026-10-27T00:00:00.000Z", "accessBundles": [], } +ON_BEHALF_OF = { + "type": "user", + "id": "0d9a7c3e-5b1f-4e2a-8c6d-7f4b2e1a9c50", + "username": "jsmith@example.com", + "email": "jsmith@example.com", + "firstName": "Jane", + "lastName": "Smith", +} @pytest.fixture @@ -82,3 +91,38 @@ def test_sessions_revoke_leaves_access_bundles_unset_when_the_response_omits_the def test_sessions_get_keeps_an_empty_access_bundles_list(api, agent_vault, json_response): api.session.get.return_value = json_response({"session": SESSION}) assert agent_vault.sessions.get(SESSION["id"]).accessBundles == [] + + +def test_sessions_create_leaves_out_ttl_so_the_server_picks_the_default(api, agent_vault, json_response): + api.session.post.return_value = json_response({"session": dict(SESSION, token="token")}) + agent_vault.sessions.create(access_bundle="release") + assert api.session.post.call_args.kwargs["json"] == {"accessBundles": ["release"]} + + +@pytest.mark.parametrize("kwargs, on_behalf_of", [ + ({"on_behalf_of_user_email": ON_BEHALF_OF["email"]}, {"userEmail": ON_BEHALF_OF["email"]}), + ({"on_behalf_of_user_id": ON_BEHALF_OF["id"]}, {"userId": ON_BEHALF_OF["id"]}), + ({"on_behalf_of_user_email": ""}, {"userEmail": ""}), +], ids=["email", "id", "empty email"]) +def test_sessions_create_names_the_user_it_acts_for(api, agent_vault, json_response, kwargs, on_behalf_of): + api.session.post.return_value = json_response({"session": dict(SESSION, token="token")}) + agent_vault.sessions.create(access_bundle="release", **kwargs) + assert api.session.post.call_args.kwargs["json"] == {"accessBundles": ["release"], "onBehalfOf": on_behalf_of} + + +def test_sessions_create_rejects_both_a_user_email_and_a_user_id(api, agent_vault): + with pytest.raises(ValueError, match="not both"): + agent_vault.sessions.create( + access_bundle="release", + on_behalf_of_user_email=ON_BEHALF_OF["email"], + on_behalf_of_user_id=ON_BEHALF_OF["id"], + ) + api.session.post.assert_not_called() + + +@pytest.mark.parametrize("on_behalf_of", [ON_BEHALF_OF, dict(ON_BEHALF_OF, id=None), None], + ids=["user", "deleted user", "own session"]) +def test_sessions_get_reads_the_user_a_session_acts_for(api, agent_vault, json_response, on_behalf_of): + api.session.get.return_value = json_response({"session": dict(SESSION, onBehalfOf=on_behalf_of)}) + expected = AgentVaultSessionActor(**on_behalf_of) if on_behalf_of else None + assert agent_vault.sessions.get(SESSION["id"]).onBehalfOf == expected