diff --git a/infisical_sdk/__init__.py b/infisical_sdk/__init__.py index 865d1e5..e2e0abf 100644 --- a/infisical_sdk/__init__.py +++ b/infisical_sdk/__init__.py @@ -1,3 +1,3 @@ from .client import InfisicalSDKClient # noqa -from .infisical_requests import InfisicalError # noqa -from .api_types import SingleSecretResponse, ListSecretsResponse, BaseSecret, SymmetricEncryption, DynamicSecretProviders, AgentVaultTrafficPolicy, AgentVaultSessionStatus, AgentVaultSessionScope # noqa \ No newline at end of file +from .infisical_requests import APIError, InfisicalError # noqa +from .api_types import SingleSecretResponse, ListSecretsResponse, BaseSecret, SymmetricEncryption, DynamicSecretProviders, AgentVaultTrafficPolicy, AgentVaultSessionStatus, AgentVaultSessionScope, AgentVaultMemberType, AgentVaultDelegationScope # noqa \ No newline at end of file diff --git a/infisical_sdk/api_types.py b/infisical_sdk/api_types.py index 7191e46..7528a35 100644 --- a/infisical_sdk/api_types.py +++ b/infisical_sdk/api_types.py @@ -450,9 +450,20 @@ def from_dict(cls, data: Dict) -> 'SingleLeaseResponse': ) class AgentVaultTrafficPolicy(str, Enum): - """Enum for which hosts an agent may reach through an Agent Vault proxy""" + """Enum for which hosts an agent may reach through an Agent Vault proxy. + + BUNDLE_HOSTS is the old name of SERVICE_HOSTS and the same member, so a proxy's traffic policy compares equal to + either one, and AgentVaultTrafficPolicy("bundle-hosts") is SERVICE_HOSTS. + """ ANY_HOST = "any-host" - BUNDLE_HOSTS = "bundle-hosts" + SERVICE_HOSTS = "service-hosts" + BUNDLE_HOSTS = "service-hosts" + + @classmethod + def _missing_(cls, value): + if value == "bundle-hosts": + return cls.SERVICE_HOSTS + return None class AgentVaultSessionStatus(str, Enum): """Enum for Agent Vault session status""" @@ -465,6 +476,17 @@ class AgentVaultSessionScope(str, Enum): MINE = "mine" ALL = "all" +class AgentVaultMemberType(str, Enum): + """Enum for the kinds of Agent Vault member: users, machine identities and groups""" + USER = "user" + MACHINE_IDENTITY = "machineIdentity" + GROUP = "group" + +class AgentVaultDelegationScope(str, Enum): + """Enum for which users a machine identity can create Agent Vault sessions on behalf of""" + ALL_MEMBERS = "all-members" + SELECTED = "selected" + @dataclass class AgentVaultProxy(BaseModel): """Infisical Agent Vault Proxy. Only admins see the settings fields, so they are None for members.""" @@ -478,6 +500,16 @@ class AgentVaultProxy(BaseModel): allowedHosts: Optional[str] = None pollInterval: Optional[int] = None createdAt: Optional[str] = None + description: Optional[str] = None + + @classmethod + def from_dict(cls, data: Dict) -> 'AgentVaultProxy': + """Create model from dictionary. An Infisical version from before service-hosts was named reports bundle-hosts, which reads as service-hosts.""" + proxy_data = data.copy() + if proxy_data.get('trafficPolicy') == 'bundle-hosts': + proxy_data['trafficPolicy'] = AgentVaultTrafficPolicy.SERVICE_HOSTS.value + + return super().from_dict(proxy_data) @dataclass class SingleAgentVaultProxyResponse(BaseModel): @@ -518,7 +550,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 @@ -533,6 +565,21 @@ class AgentVaultSessionAccessBundle(BaseModel): name: str id: Optional[str] = None +@dataclass +class AgentVaultInlineService(BaseModel): + """A service sent with an Agent Vault session, without its token, password or values. Its nested settings, such as credential, are left as the API returns them.""" + id: str + name: str + type: str + hostPattern: str + serverUrl: Optional[str] = None + transport: Optional[str] = None + allowedMethods: Optional[List[str]] = None + allowedPathPrefixes: Optional[List[str]] = None + credential: Optional[Dict[str, Any]] = None + customHeaders: Optional[List[Dict[str, Any]]] = None + substitutions: Optional[List[Dict[str, Any]]] = None + @dataclass class AgentVaultSession(BaseModel): """Infisical Agent Vault Session""" @@ -544,15 +591,21 @@ class AgentVaultSession(BaseModel): expiresAt: Optional[str] = None revokedAt: Optional[str] = None accessBundle: Optional[AgentVaultSessionAccessBundle] = None + onBehalfOf: Optional[AgentVaultSessionActor] = None + services: Optional[List[AgentVaultInlineService]] = None @classmethod def from_dict(cls, data: Dict) -> 'AgentVaultSession': - """Create model from dictionary with nested actor and access bundle. accessBundle is None if the response leaves it out.""" + """Create model from dictionary with nested actor, on-behalf-of user, access bundle and inline services. accessBundle and services are None if the response leaves them 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('accessBundle') is not None: session_data['accessBundle'] = AgentVaultSessionAccessBundle.from_dict(data['accessBundle']) + if data.get('services') is not None: + session_data['services'] = [AgentVaultInlineService.from_dict(service) for service in data['services']] return super().from_dict(session_data) @@ -565,13 +618,17 @@ class CreatedAgentVaultSession(BaseModel): label: Optional[str] = None expiresAt: Optional[str] = None accessBundle: Optional[AgentVaultSessionAccessBundle] = None + services: List[AgentVaultInlineService] = field(default_factory=list) @classmethod def from_dict(cls, data: Dict) -> 'CreatedAgentVaultSession': - """Create model from dictionary with nested access bundle""" + """Create model from dictionary with nested access bundle and inline services""" session_data = data.copy() if data.get('accessBundle') is not None: session_data['accessBundle'] = AgentVaultSessionAccessBundle.from_dict(data['accessBundle']) + session_data['services'] = [ + AgentVaultInlineService.from_dict(service) for service in data.get('services', []) + ] return super().from_dict(session_data) @@ -609,3 +666,248 @@ def from_dict(cls, data: Dict) -> 'ListAgentVaultSessionsResponse': sessions=[AgentVaultSession.from_dict(session) for session in data['sessions']], totalCount=data['totalCount'], ) + +@dataclass +class AgentVaultServiceRef(BaseModel): + """An Agent Vault service, by ID and name""" + id: str + name: str + +@dataclass +class AgentVaultAccessBundleRef(BaseModel): + """An Agent Vault access bundle, by ID and name""" + id: str + name: str + +@dataclass +class AgentVaultService(BaseModel): + """A service in an Agent Vault access bundle. Its nested settings, such as credential, are left as the API returns them.""" + id: str + accessBundleId: str + name: str + type: str + hostPattern: str + serverUrl: Optional[str] = None + transport: Optional[str] = None + allowedMethods: Optional[List[str]] = None + allowedPathPrefixes: Optional[List[str]] = None + credential: Optional[Dict[str, Any]] = None + customHeaders: Optional[List[Dict[str, Any]]] = None + substitutions: Optional[List[Dict[str, Any]]] = None + variableReferences: Optional[List[Dict[str, Any]]] = None + personalValues: Optional[Dict[str, Any]] = None + createdAt: Optional[str] = None + updatedAt: Optional[str] = None + allowedTools: Optional[List[str]] = None + +@dataclass +class AgentVaultAccessBundle(BaseModel): + """Infisical Agent Vault Access Bundle. Only list sets the counts, hostPatterns and servicesRequiringSetup.""" + id: str + name: str + description: Optional[str] = None + createdAt: Optional[str] = None + updatedAt: Optional[str] = None + serviceCount: Optional[int] = None + memberCount: Optional[int] = None + hostPatterns: Optional[List[str]] = None + servicesRequiringSetup: Optional[List[AgentVaultServiceRef]] = None + services: Optional[List[AgentVaultService]] = None + + @classmethod + def from_dict(cls, data: Dict) -> 'AgentVaultAccessBundle': + """Create model from dictionary with nested services""" + bundle_data = data.copy() + if data.get('servicesRequiringSetup') is not None: + bundle_data['servicesRequiringSetup'] = [ + AgentVaultServiceRef.from_dict(service) for service in data['servicesRequiringSetup'] + ] + if data.get('services') is not None: + bundle_data['services'] = [AgentVaultService.from_dict(service) for service in data['services']] + + return super().from_dict(bundle_data) + +@dataclass +class SingleAgentVaultAccessBundleResponse(BaseModel): + """Response model for get/create/update/delete Agent Vault access bundle API""" + accessBundle: AgentVaultAccessBundle + + @classmethod + def from_dict(cls, data: Dict) -> 'SingleAgentVaultAccessBundleResponse': + return cls( + accessBundle=AgentVaultAccessBundle.from_dict(data['accessBundle']), + ) + +@dataclass +class ListAgentVaultAccessBundlesResponse(BaseModel): + """Response model for list Agent Vault access bundles API""" + accessBundles: List[AgentVaultAccessBundle] + totalCount: int + + @classmethod + def from_dict(cls, data: Dict) -> 'ListAgentVaultAccessBundlesResponse': + return cls( + accessBundles=[AgentVaultAccessBundle.from_dict(bundle) for bundle in data['accessBundles']], + totalCount=data['totalCount'], + ) + +@dataclass +class AgentVaultGrantee(BaseModel): + """A user, machine identity or group that an Agent Vault access bundle is granted to""" + type: str + id: str + grantedAt: Optional[str] = None + username: Optional[str] = None + email: Optional[str] = None + firstName: Optional[str] = None + lastName: Optional[str] = None + name: Optional[str] = None + isOrgMembershipPending: Optional[bool] = None + +@dataclass +class ListAgentVaultGranteesResponse(BaseModel): + """Response model for list Agent Vault access bundle grantees API""" + members: List[AgentVaultGrantee] + totalCount: int + + @classmethod + def from_dict(cls, data: Dict) -> 'ListAgentVaultGranteesResponse': + return cls( + members=[AgentVaultGrantee.from_dict(member) for member in data['members']], + totalCount=data['totalCount'], + ) + +@dataclass +class AgentVaultGrantChangeResponse(BaseModel): + """Response model for grant/revoke Agent Vault access bundle API. Its grantees have only type, id and grantedAt.""" + members: List[AgentVaultGrantee] + skipped: List[AgentVaultGrantee] + + @classmethod + def from_dict(cls, data: Dict) -> 'AgentVaultGrantChangeResponse': + return cls( + members=[AgentVaultGrantee.from_dict(member) for member in data['members']], + skipped=[AgentVaultGrantee.from_dict(member) for member in data['skipped']], + ) + +@dataclass +class AgentVaultMember(BaseModel): + """A user, machine identity or group that's a member of Agent Vault, with its Agent Vault role""" + type: str + id: str + role: Optional[str] = None + isActive: Optional[bool] = None + addedAt: Optional[str] = None + username: Optional[str] = None + email: Optional[str] = None + firstName: Optional[str] = None + lastName: Optional[str] = None + isOrgMembershipPending: Optional[bool] = None + name: Optional[str] = None + isManagedByAgentVault: Optional[bool] = None + orgId: Optional[str] = None + +@dataclass +class ListAgentVaultMembersResponse(BaseModel): + """Response model for list available Agent Vault grantees and delegation subjects API""" + members: List[AgentVaultMember] + totalCount: int + + @classmethod + def from_dict(cls, data: Dict) -> 'ListAgentVaultMembersResponse': + return cls( + members=[AgentVaultMember.from_dict(member) for member in data['members']], + totalCount=data['totalCount'], + ) + +def _missing_access_bundles(data: Dict) -> Optional[List[AgentVaultAccessBundleRef]]: + """Parses missingAccessBundles, which only some delegation responses include, so it's None when left out.""" + bundles = data.get('missingAccessBundles') + return None if bundles is None else [AgentVaultAccessBundleRef.from_dict(bundle) for bundle in bundles] + +@dataclass +class AgentVaultDelegationMachineIdentity(BaseModel): + """The machine identity an Agent Vault delegation belongs to""" + id: str + name: str + +@dataclass +class AgentVaultDelegationUser(BaseModel): + """A user in an Agent Vault delegation. missingAccessBundles is only set by get, set, add_subjects and remove_subjects.""" + id: str + username: str + addedAt: str + email: Optional[str] = None + firstName: Optional[str] = None + lastName: Optional[str] = None + missingAccessBundles: Optional[List[AgentVaultAccessBundleRef]] = None + + @classmethod + def from_dict(cls, data: Dict) -> 'AgentVaultDelegationUser': + """Create model from dictionary with nested missing access bundles""" + user_data = data.copy() + user_data['missingAccessBundles'] = _missing_access_bundles(data) + + return super().from_dict(user_data) + +@dataclass +class AgentVaultDelegationGroup(BaseModel): + """A group in an Agent Vault delegation. missingAccessBundles is only set by get, set, add_subjects and remove_subjects.""" + id: str + name: str + addedAt: str + missingAccessBundles: Optional[List[AgentVaultAccessBundleRef]] = None + + @classmethod + def from_dict(cls, data: Dict) -> 'AgentVaultDelegationGroup': + """Create model from dictionary with nested missing access bundles""" + group_data = data.copy() + group_data['missingAccessBundles'] = _missing_access_bundles(data) + + return super().from_dict(group_data) + +@dataclass +class AgentVaultDelegation(BaseModel): + """Infisical Agent Vault Delegation: which users a machine identity can create sessions on behalf of""" + machineIdentity: AgentVaultDelegationMachineIdentity + scope: str + users: List[AgentVaultDelegationUser] + groups: List[AgentVaultDelegationGroup] + createdAt: str + updatedAt: str + missingAccessBundles: Optional[List[AgentVaultAccessBundleRef]] = None + + @classmethod + def from_dict(cls, data: Dict) -> 'AgentVaultDelegation': + """Create model from dictionary with nested machine identity, users, groups and missing access bundles""" + delegation_data = data.copy() + delegation_data['machineIdentity'] = AgentVaultDelegationMachineIdentity.from_dict(data['machineIdentity']) + delegation_data['users'] = [AgentVaultDelegationUser.from_dict(user) for user in data['users']] + delegation_data['groups'] = [AgentVaultDelegationGroup.from_dict(group) for group in data['groups']] + delegation_data['missingAccessBundles'] = _missing_access_bundles(data) + + return super().from_dict(delegation_data) + +@dataclass +class SingleAgentVaultDelegationResponse(BaseModel): + """Response model for Agent Vault delegation API. delegation is None when remove_subjects ends the delegation.""" + delegation: Optional[AgentVaultDelegation] = None + + @classmethod + def from_dict(cls, data: Dict) -> 'SingleAgentVaultDelegationResponse': + return cls( + delegation=AgentVaultDelegation.from_dict(data['delegation']) if data.get('delegation') is not None else None, + ) + +@dataclass +class ListAgentVaultDelegationsResponse(BaseModel): + """Response model for list Agent Vault delegations API""" + delegations: List[AgentVaultDelegation] + totalCount: int + + @classmethod + def from_dict(cls, data: Dict) -> 'ListAgentVaultDelegationsResponse': + return cls( + delegations=[AgentVaultDelegation.from_dict(delegation) for delegation in data['delegations']], + totalCount=data['totalCount'], + ) diff --git a/infisical_sdk/infisical_requests.py b/infisical_sdk/infisical_requests.py index b50bfc0..5d75363 100644 --- a/infisical_sdk/infisical_requests.py +++ b/infisical_sdk/infisical_requests.py @@ -256,6 +256,31 @@ def patch( headers=dict(response.headers) ) + @with_retry(max_retries=4, base_delay=1.0) + def put( + self, + path: str, + model: Type[T], + json: Optional[Dict[str, Any]] = None + ) -> APIResponse[T]: + + """Make a PUT request with JSON data. A PUT replaces the whole resource, so it's safe to retry.""" + + if json is not None: + # Filter out None values + json = {k: v for k, v in json.items() if v is not None} + + response = self.session.put(self._build_url(path), json=json) + data = self._handle_response(response) + + parsed_data = model.from_dict(data) if hasattr(model, 'from_dict') else data + + return APIResponse( + data=parsed_data, + status_code=response.status_code, + headers=dict(response.headers) + ) + @with_retry(max_retries=4, base_delay=1.0) def delete( self, @@ -264,7 +289,32 @@ def delete( json: Optional[Dict[str, Any]] = None ) -> APIResponse[T]: - """Make a PATCH request with JSON data""" + """Make a DELETE request with JSON data""" + + if json is not None: + # Filter out None values + json = {k: v for k, v in json.items() if v is not None} + + response = self.session.delete(self._build_url(path), json=json) + data = self._handle_response(response) + + parsed_data = model.from_dict(data) if hasattr(model, 'from_dict') else data + + return APIResponse( + data=parsed_data, + status_code=response.status_code, + headers=dict(response.headers) + ) + + @with_retry(max_retries=4, base_delay=1.0, network_errors=CONNECT_ERRORS) + def delete_once( + self, + path: str, + model: Type[T], + json: Optional[Dict[str, Any]] = None + ) -> APIResponse[T]: + + """Make a DELETE request that must not run twice. It's retried only in the same cases as post_once.""" if json is not None: # Filter out None values diff --git a/infisical_sdk/resources/agent_vault.py b/infisical_sdk/resources/agent_vault.py index d727ab6..5ebc411 100644 --- a/infisical_sdk/resources/agent_vault.py +++ b/infisical_sdk/resources/agent_vault.py @@ -1,29 +1,81 @@ -from typing import Optional, List, Union +from typing import Any, Dict, Optional, List, Union +from urllib.parse import quote from infisical_sdk.infisical_requests import InfisicalRequests from infisical_sdk.api_types import ( + AgentVaultAccessBundle, + AgentVaultDelegation, + AgentVaultDelegationScope, + AgentVaultMemberType, AgentVaultProxy, AgentVaultSession, AgentVaultSessionScope, AgentVaultSessionStatus, AgentVaultTrafficPolicy, + AgentVaultGrantChangeResponse, AgentVaultProxyWithTokenResponse, CreateAgentVaultSessionResponse, CreatedAgentVaultSession, + ListAgentVaultAccessBundlesResponse, + ListAgentVaultDelegationsResponse, + ListAgentVaultGranteesResponse, + ListAgentVaultMembersResponse, ListAgentVaultProxiesResponse, ListAgentVaultSessionsResponse, + SingleAgentVaultAccessBundleResponse, + SingleAgentVaultDelegationResponse, SingleAgentVaultProxyResponse, SingleAgentVaultSessionResponse, ) +def _segment(identifier: Any) -> str: + """Encodes an ID as one path segment. The URL would resolve "." and ".." like directories, so a revoke with ".." + would reach the route that removes members from Agent Vault, which takes the same body. An empty ID would reach a + route that isn't the one meant either, so all three are refused before any request.""" + value = "" if identifier is None else str(identifier) + if value in ("", ".", ".."): + raise ValueError(f'Invalid ID {value!r}: an ID can\'t be empty, "." or ".."') + return quote(value, safe="") + + def _join_hosts(allowed_hosts: Optional[Union[str, List[str]]]) -> Optional[str]: - """The API takes host patterns as one comma-separated string, and rejects an empty one.""" + """The API takes host patterns as one comma-separated string and rejects an empty one, so empty values become None.""" if isinstance(allowed_hosts, list): - return ",".join(allowed_hosts) or None + allowed_hosts = ",".join(allowed_hosts) + if allowed_hosts is not None and not allowed_hosts.strip(): + return None return allowed_hosts +def _query_bool(value: Optional[bool]) -> Optional[str]: + """The API reads a query boolean only as "true" or "false", and requests would send True as "True".""" + if value is None: + return None + return "true" if value else "false" + + +def _wire_traffic_policy(traffic_policy: Optional[Union[AgentVaultTrafficPolicy, str]]) -> Optional[str]: + """Infisical stores service-hosts under its old name, bundle-hosts, and still accepts that name, while a version + from before the rename accepts only bundle-hosts. So either name goes out as bundle-hosts, which every version takes.""" + if isinstance(traffic_policy, AgentVaultTrafficPolicy): + traffic_policy = traffic_policy.value + return "bundle-hosts" if traffic_policy == AgentVaultTrafficPolicy.SERVICE_HOSTS.value else traffic_policy + + +def _with_services(session: AgentVaultSession) -> AgentVaultSession: + """A session from an Infisical version without inline services has none, so it gets an empty list rather than None.""" + if session.services is None: + session.services = [] + return session + + +def _check_one_user(on_behalf_of_user_email: Optional[str], on_behalf_of_user_id: Optional[str]) -> None: + """The API names a user by email address or by ID, not both.""" + 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") + + class AgentVaultProxies: """Manages Agent Vault proxies.""" @@ -40,7 +92,7 @@ def list( """List the organization's Agent Vault proxies. Args: - search: Match proxies by name. + search: Match proxies by name or description. order_by: What to sort proxies by: "name" or "createdAt". Defaults to "createdAt". order_direction: Which way to sort: "asc" or "desc". Defaults to "desc". limit: The maximum number of proxies to return (1-100). Defaults to 20. @@ -70,23 +122,26 @@ def create( name: str, traffic_policy: Union[AgentVaultTrafficPolicy, str] = AgentVaultTrafficPolicy.ANY_HOST, allowed_hosts: Optional[Union[str, List[str]]] = None, - poll_interval: int = 60) -> AgentVaultProxyWithTokenResponse: + poll_interval: int = 60, + description: Optional[str] = None) -> AgentVaultProxyWithTokenResponse: """Register an Agent Vault proxy and issue its proxy token. Args: name: The name of the proxy (a slug, at most 64 characters). - traffic_policy: Which hosts an agent may reach through this proxy. Defaults to "any-host". - allowed_hosts: Hosts that stay reachable under the "bundle-hosts" traffic policy even though no access bundle covers them. A list or a comma-separated string. + traffic_policy: Which hosts an agent may reach through this proxy: AgentVaultTrafficPolicy.ANY_HOST ("any-host"), or AgentVaultTrafficPolicy.SERVICE_HOSTS ("service-hosts") for only the hosts that the session's services cover. Defaults to "any-host". The old name of SERVICE_HOSTS, BUNDLE_HOSTS ("bundle-hosts"), works the same. + allowed_hosts: Hosts that stay reachable under the "service-hosts" traffic policy even though none of the session's services cover them. A list or a comma-separated string. poll_interval: How often, in seconds, the proxy refreshes its sessions and settings (10-300). Defaults to 60. + description: What the proxy is for, up to 256 characters, without line breaks or other control characters. Returns: AgentVaultProxyWithTokenResponse containing the proxy and its proxyToken. The token is only returned once. """ request_body = { "name": name, - "trafficPolicy": traffic_policy.value if isinstance(traffic_policy, AgentVaultTrafficPolicy) else traffic_policy, + "trafficPolicy": _wire_traffic_policy(traffic_policy), "allowedHosts": _join_hosts(allowed_hosts), "pollInterval": poll_interval, + "description": description, } result = self.requests.post_once( @@ -104,40 +159,37 @@ def update( traffic_policy: Optional[Union[AgentVaultTrafficPolicy, str]] = None, allowed_hosts: Optional[Union[str, List[str]]] = None, poll_interval: Optional[int] = None, - clear_allowed_hosts: bool = False) -> AgentVaultProxy: - """Update an Agent Vault proxy's name or settings. Only the values you pass are changed. + description: Optional[str] = None) -> AgentVaultProxy: + """Update an Agent Vault proxy's name, description or settings. Only the values you pass are changed. Args: proxy_id: The ID of the proxy to update. name: The new name of the proxy. - traffic_policy: Which hosts an agent may reach through this proxy. - allowed_hosts: Hosts that stay reachable under the "bundle-hosts" traffic policy. A list or a comma-separated string. + traffic_policy: Which hosts an agent may reach through this proxy: AgentVaultTrafficPolicy.ANY_HOST ("any-host"), or AgentVaultTrafficPolicy.SERVICE_HOSTS ("service-hosts") for only the hosts that the session's services cover. The old name of SERVICE_HOSTS, BUNDLE_HOSTS ("bundle-hosts"), works the same. + allowed_hosts: Hosts that stay reachable under the "service-hosts" traffic policy. A list or a comma-separated string. Pass an empty list or string to remove every allowed host. poll_interval: How often, in seconds, the proxy refreshes its sessions and settings (10-300). - clear_allowed_hosts: Remove all allowed hosts. Can't be combined with allowed_hosts. + description: The new description, up to 256 characters, without line breaks or other control characters. Pass an empty string to remove the description. Returns: The updated proxy. """ - if clear_allowed_hosts and allowed_hosts is not None: - raise ValueError("Pass either allowed_hosts or clear_allowed_hosts, not both") - if allowed_hosts == []: - raise ValueError("allowed_hosts can't be empty. To remove every allowed host, pass clear_allowed_hosts=True") - - if isinstance(traffic_policy, AgentVaultTrafficPolicy): - traffic_policy = traffic_policy.value + hosts = _join_hosts(allowed_hosts) + # An empty value removes every allowed host, which the API takes as null. + remove_hosts = allowed_hosts is not None and hosts is None request_body = { "name": name, - "trafficPolicy": traffic_policy, - "allowedHosts": _join_hosts(allowed_hosts), + "trafficPolicy": _wire_traffic_policy(traffic_policy), + "allowedHosts": hosts, "pollInterval": poll_interval, + "description": description, } result = self.requests.patch( - path=f"/api/v1/agent-vault/proxies/{proxy_id}", + path=f"/api/v1/agent-vault/proxies/{_segment(proxy_id)}", json=request_body, model=SingleAgentVaultProxyResponse, - nullable_keys=["allowedHosts"] if clear_allowed_hosts else None + nullable_keys=["allowedHosts"] if remove_hosts else None ) return result.data.proxy @@ -151,8 +203,10 @@ def delete(self, proxy_id: str) -> AgentVaultProxy: Returns: The deleted proxy. Only id and name are set. """ - result = self.requests.delete( - path=f"/api/v1/agent-vault/proxies/{proxy_id}", + # Not retried once the request may have reached the server: after a lost response, a retry would fail with a + # 404 for a delete that worked. + result = self.requests.delete_once( + path=f"/api/v1/agent-vault/proxies/{_segment(proxy_id)}", model=SingleAgentVaultProxyResponse ) @@ -168,7 +222,7 @@ def rotate_token(self, proxy_id: str) -> AgentVaultProxyWithTokenResponse: AgentVaultProxyWithTokenResponse containing the proxy and its new proxyToken. The token is only returned once. """ result = self.requests.post_once( - path=f"/api/v1/agent-vault/proxies/{proxy_id}/token-auth/proxy-token", + path=f"/api/v1/agent-vault/proxies/{_segment(proxy_id)}/token-auth/proxy-token", model=AgentVaultProxyWithTokenResponse ) @@ -184,7 +238,7 @@ def revoke(self, proxy_id: str) -> AgentVaultProxy: The revoked proxy. """ result = self.requests.post( - path=f"/api/v1/agent-vault/proxies/{proxy_id}/revoke", + path=f"/api/v1/agent-vault/proxies/{_segment(proxy_id)}/revoke", model=SingleAgentVaultProxyResponse ) @@ -209,7 +263,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. @@ -234,6 +288,8 @@ def list( model=ListAgentVaultSessionsResponse ) + for session in result.data.sessions: + _with_services(session) return result.data def get(self, session_id: str) -> AgentVaultSession: @@ -246,31 +302,56 @@ def get(self, session_id: str) -> AgentVaultSession: The session. """ result = self.requests.get( - path=f"/api/v1/agent-vault/sessions/{session_id}", + path=f"/api/v1/agent-vault/sessions/{_segment(session_id)}", model=SingleAgentVaultSessionResponse ) - return result.data.session + return _with_services(result.data.session) 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. + access_bundle: Optional[str] = None, + ttl: Optional[str] = None, + label: Optional[str] = None, + on_behalf_of_user_email: Optional[str] = None, + on_behalf_of_user_id: Optional[str] = None, + services: Optional[List[Dict[str, Any]]] = None) -> CreatedAgentVaultSession: + """Create an Agent Vault session with an access bundle you can reach, services of its own, or both, for yourself or, as a machine identity with a delegation, on behalf of a user. + + Only Agent Vault admins can send services with a session. On behalf of a user, that's your machine identity: + the user doesn't need the Admin role, and the services don't use the user's personal values. + + 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 unless + they're an Agent Vault admin. 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". + access_bundle: The name of the access bundle the session carries. Pass access_bundle, services, or both. + 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. + services: Services the session carries itself, instead of or beside its access bundle's: up to 20 dicts with the API's field names, such as {"type": "http", "hostPattern": "api.github.com", "credential": {"type": "bearer", "value": token}}. Only Agent Vault admins can send them. Each has the same fields as a service in an access bundle, except that its name is optional, its values can't use variables, it can't use OAuth or member-provided values, and an MCP server can't be limited to some of its tools. To limit an MCP server's tools, add it to an access bundle instead. Returns: - The created session, including its token. The token is only returned once. + The created session, including its token. The token is only returned once. Its services leave out their tokens, passwords and values. """ + _check_one_user(on_behalf_of_user_email, on_behalf_of_user_id) + + 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 = { + # Left out without an access bundle, for a session that carries only services of its own. "accessBundle": access_bundle, + "services": services, "ttl": ttl, "label": label, + "onBehalfOf": on_behalf_of, } result = self.requests.post_once( @@ -286,7 +367,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. @@ -296,13 +377,13 @@ def update(self, session_id: str, label: Optional[str]) -> AgentVaultSession: } result = self.requests.patch( - path=f"/api/v1/agent-vault/sessions/{session_id}", + path=f"/api/v1/agent-vault/sessions/{_segment(session_id)}", json=request_body, model=SingleAgentVaultSessionResponse, nullable_keys=["label"] ) - return result.data.session + return _with_services(result.data.session) def revoke(self, session_id: str) -> AgentVaultSession: """Revoke an Agent Vault session. Revoking an already revoked session is a no-op. @@ -311,16 +392,484 @@ def revoke(self, session_id: str) -> AgentVaultSession: session_id: The ID of the session to revoke. Returns: - The revoked session. Its accessBundle is None, because the response leaves the bundle out; call get for it. + The revoked session. Its accessBundle and services are None, because the response leaves them out; call get for them. """ result = self.requests.post( - path=f"/api/v1/agent-vault/sessions/{session_id}/revoke", + path=f"/api/v1/agent-vault/sessions/{_segment(session_id)}/revoke", model=SingleAgentVaultSessionResponse ) return result.data.session +class AgentVaultAccessBundles: + """Manages Agent Vault access bundles and who they're granted to. Every method except list and get requires the Admin role in Agent Vault.""" + + def __init__(self, requests: InfisicalRequests) -> None: + self.requests = requests + + def list( + self, + search: Optional[str] = None, + requires_setup: Optional[bool] = None, + granted: Optional[bool] = None, + on_behalf_of_user_email: Optional[str] = None, + on_behalf_of_user_id: Optional[str] = None, + order_by: str = "createdAt", + order_direction: str = "desc", + limit: int = 20, + offset: int = 0) -> ListAgentVaultAccessBundlesResponse: + """List the Agent Vault access bundles you can reach. Admins reach every access bundle. + + Args: + search: Match access bundles by name or description. + requires_setup: True lists only the access bundles with services that still need your personal values, and False lists the rest. Defaults to both. + granted: True lists only the access bundles granted to you, directly or through a group, and False lists the ones you reach only as an admin. Defaults to both. Can't be combined with on_behalf_of_user_email or on_behalf_of_user_id. + on_behalf_of_user_email: As a machine identity whose delegation covers the user with this email address, list the access bundles that a session on the user's behalf could use: those both the user and your machine identity can use. requires_setup and servicesRequiringSetup then refer to the user's personal values. Can't be combined with on_behalf_of_user_id. + on_behalf_of_user_id: Like on_behalf_of_user_email, naming the user by ID instead. + order_by: What to sort access bundles by: "name", "serviceCount" or "createdAt". Defaults to "createdAt". + order_direction: Which way to sort: "asc" or "desc". Defaults to "desc". + limit: The maximum number of access bundles to return (1-100). Defaults to 20. + offset: How many access bundles to skip. Defaults to 0. + + Returns: + ListAgentVaultAccessBundlesResponse containing accessBundles and totalCount. + """ + _check_one_user(on_behalf_of_user_email, on_behalf_of_user_id) + + params = { + "search": search, + "requiresSetup": _query_bool(requires_setup), + "granted": _query_bool(granted), + "onBehalfOfUserEmail": on_behalf_of_user_email, + "onBehalfOfUserId": on_behalf_of_user_id, + "orderBy": order_by, + "orderDirection": order_direction, + "limit": limit, + "offset": offset, + } + + result = self.requests.get( + path="/api/v1/agent-vault/access-bundles", + params=params, + model=ListAgentVaultAccessBundlesResponse + ) + + return result.data + + def get(self, access_bundle_id: str) -> AgentVaultAccessBundle: + """Get an Agent Vault access bundle you can reach, with its services. + + Args: + access_bundle_id: The ID of the access bundle. + + Returns: + The access bundle, with its services in services. A service's variableReferences is only set for admins. + """ + result = self.requests.get( + path=f"/api/v1/agent-vault/access-bundles/{_segment(access_bundle_id)}", + model=SingleAgentVaultAccessBundleResponse + ) + + return result.data.accessBundle + + def create(self, name: str, description: Optional[str] = None) -> AgentVaultAccessBundle: + """Create an Agent Vault access bundle. It's granted to you, unless you're in Agent Vault only through a group. + + Args: + name: The name of the access bundle (a slug, at most 64 characters). + description: What the access bundle is for, up to 256 characters, without line breaks or other control characters. + + Returns: + The created access bundle. + """ + request_body = { + "name": name, + "description": description, + } + + result = self.requests.post_once( + path="/api/v1/agent-vault/access-bundles", + json=request_body, + model=SingleAgentVaultAccessBundleResponse + ) + + return result.data.accessBundle + + def update( + self, + access_bundle_id: str, + name: Optional[str] = None, + description: Optional[str] = None) -> AgentVaultAccessBundle: + """Update an Agent Vault access bundle's name or description. Only the values you pass are changed. + + Args: + access_bundle_id: The ID of the access bundle to update. + name: The new name of the access bundle (a slug, at most 64 characters). + description: The new description, up to 256 characters, without line breaks or other control characters. Pass an empty string to remove the description. + + Returns: + The updated access bundle. + """ + request_body = { + "name": name, + "description": description, + } + + result = self.requests.patch( + path=f"/api/v1/agent-vault/access-bundles/{_segment(access_bundle_id)}", + json=request_body, + model=SingleAgentVaultAccessBundleResponse + ) + + return result.data.accessBundle + + def delete(self, access_bundle_id: str) -> AgentVaultAccessBundle: + """Delete an Agent Vault access bundle. + + Args: + access_bundle_id: The ID of the access bundle to delete. + + Returns: + The deleted access bundle. + """ + # Not retried once the request may have reached the server: after a lost response, a retry would fail with a + # 404 for a delete that worked. + result = self.requests.delete_once( + path=f"/api/v1/agent-vault/access-bundles/{_segment(access_bundle_id)}", + model=SingleAgentVaultAccessBundleResponse + ) + + return result.data.accessBundle + + def list_grantees( + self, + access_bundle_id: str, + actor_type: Optional[Union[AgentVaultMemberType, str]] = None, + search: Optional[str] = None, + limit: int = 20, + offset: int = 0) -> ListAgentVaultGranteesResponse: + """List the users, machine identities and groups that an Agent Vault access bundle is granted to. + + Args: + access_bundle_id: The ID of the access bundle. + actor_type: List only users (AgentVaultMemberType.USER), only machine identities (AgentVaultMemberType.MACHINE_IDENTITY), or only groups (AgentVaultMemberType.GROUP). Defaults to all three. + search: Match grantees by name, username or email address. + limit: The maximum number of grantees to return (1-100). Defaults to 20. + offset: How many grantees to skip. Defaults to 0. + + Returns: + ListAgentVaultGranteesResponse containing the grantees in members, and totalCount. + """ + params = { + "actorType": actor_type.value if isinstance(actor_type, AgentVaultMemberType) else actor_type, + "search": search, + "limit": limit, + "offset": offset, + } + + result = self.requests.get( + path=f"/api/v1/agent-vault/access-bundles/{_segment(access_bundle_id)}/members", + params=params, + model=ListAgentVaultGranteesResponse + ) + + return result.data + + def list_available_grantees( + self, + access_bundle_id: str, + actor_type: Optional[Union[AgentVaultMemberType, str]] = None, + search: Optional[str] = None, + limit: int = 20, + offset: int = 0) -> ListAgentVaultMembersResponse: + """List the Agent Vault members that an access bundle isn't granted to yet, so you can grant it to them. + + Args: + access_bundle_id: The ID of the access bundle. + actor_type: List only users (AgentVaultMemberType.USER), only machine identities (AgentVaultMemberType.MACHINE_IDENTITY), or only groups (AgentVaultMemberType.GROUP). Defaults to all three. + search: Match members by name, username or email address. + limit: The maximum number of members to return (1-100). Defaults to 20. + offset: How many members to skip. Defaults to 0. + + Returns: + ListAgentVaultMembersResponse containing members and totalCount. + """ + params = { + "actorType": actor_type.value if isinstance(actor_type, AgentVaultMemberType) else actor_type, + "search": search, + "limit": limit, + "offset": offset, + } + + result = self.requests.get( + path=f"/api/v1/agent-vault/access-bundles/{_segment(access_bundle_id)}/members/available", + params=params, + model=ListAgentVaultMembersResponse + ) + + return result.data + + def grant( + self, + access_bundle_id: str, + user_ids: Optional[List[str]] = None, + machine_identity_ids: Optional[List[str]] = None, + group_ids: Optional[List[str]] = None) -> AgentVaultGrantChangeResponse: + """Grant an Agent Vault access bundle to users, machine identities or groups. Each must be a member of Agent Vault. Name at least one, and at most 100 in total. + + Args: + access_bundle_id: The ID of the access bundle. + user_ids: The IDs of the users to grant the access bundle to. + machine_identity_ids: The IDs of the machine identities to grant the access bundle to. + group_ids: The IDs of the groups to grant the access bundle to. + + Returns: + AgentVaultGrantChangeResponse with the new grants in members, and the grantees that already had the access bundle in skipped. + """ + request_body = { + "userIds": user_ids, + "machineIdentityIds": machine_identity_ids, + "groupIds": group_ids, + } + + # Not retried once the request may have reached the server: after a lost response, a retry would report + # each grantee as skipped. + result = self.requests.post_once( + path=f"/api/v1/agent-vault/access-bundles/{_segment(access_bundle_id)}/members", + json=request_body, + model=AgentVaultGrantChangeResponse + ) + + return result.data + + def revoke( + self, + access_bundle_id: str, + user_ids: Optional[List[str]] = None, + machine_identity_ids: Optional[List[str]] = None, + group_ids: Optional[List[str]] = None) -> AgentVaultGrantChangeResponse: + """Revoke an Agent Vault access bundle from users, machine identities or groups. Name at least one, and at most 100 in total. + + Args: + access_bundle_id: The ID of the access bundle. + user_ids: The IDs of the users to revoke the access bundle from. + machine_identity_ids: The IDs of the machine identities to revoke the access bundle from. + group_ids: The IDs of the groups to revoke the access bundle from. + + Returns: + AgentVaultGrantChangeResponse with the revoked grants in members, and the grantees that didn't have the access bundle in skipped. + """ + request_body = { + "userIds": user_ids, + "machineIdentityIds": machine_identity_ids, + "groupIds": group_ids, + } + + # Not retried once the request may have reached the server: after a lost response, a retry would report + # each grantee as skipped. + result = self.requests.delete_once( + path=f"/api/v1/agent-vault/access-bundles/{_segment(access_bundle_id)}/members", + json=request_body, + model=AgentVaultGrantChangeResponse + ) + + return result.data + + +class AgentVaultDelegations: + """Manages Agent Vault delegations, which let machine identities create sessions on behalf of users. Every method requires the Admin role in Agent Vault, except that a machine identity can get its own delegation.""" + + def __init__(self, requests: InfisicalRequests) -> None: + self.requests = requests + + def list( + self, + machine_identity_ids: Optional[List[str]] = None, + limit: int = 100, + offset: int = 0) -> ListAgentVaultDelegationsResponse: + """List the machine identities that can create Agent Vault sessions on behalf of users. + + Args: + machine_identity_ids: Only return the delegations of these machine identities, up to 100. + limit: The maximum number of delegations to return (1-100). Defaults to 100. + offset: How many delegations to skip. Defaults to 0. + + Returns: + ListAgentVaultDelegationsResponse containing delegations and totalCount. missingAccessBundles isn't set on these delegations; call get for it. + """ + params = { + "machineIdentityIds": ",".join(machine_identity_ids) if machine_identity_ids else None, + "limit": limit, + "offset": offset, + } + + result = self.requests.get( + path="/api/v1/agent-vault/delegations", + params=params, + model=ListAgentVaultDelegationsResponse + ) + + return result.data + + def get(self, machine_identity_id: str) -> AgentVaultDelegation: + """Get which users a machine identity can create Agent Vault sessions on behalf of. Raises an APIError with status 404 if the machine identity has no delegation. + + Args: + machine_identity_id: The ID of the machine identity. + + Returns: + The delegation. missingAccessBundles lists the access bundles that the users it covers can use and the machine identity can't, so the machine identity can't create sessions over them. They're empty when a machine identity that isn't an admin gets its own delegation. + """ + result = self.requests.get( + path=f"/api/v1/agent-vault/delegations/{_segment(machine_identity_id)}", + model=SingleAgentVaultDelegationResponse + ) + + return result.data.delegation + + def set( + self, + machine_identity_id: str, + scope: Union[AgentVaultDelegationScope, str], + user_ids: Optional[List[str]] = None, + group_ids: Optional[List[str]] = None) -> AgentVaultDelegation: + """Set which users a machine identity can create Agent Vault sessions on behalf of, replacing any delegation it has. + + Args: + machine_identity_id: The ID of the machine identity. It must be a member of Agent Vault. + scope: AgentVaultDelegationScope.ALL_MEMBERS ("all-members") for every Agent Vault member, admins included, or AgentVaultDelegationScope.SELECTED ("selected") for only the users and groups you name. + user_ids: With the "selected" scope, the users the machine identity can act on behalf of, up to 100. Each must be a member of Agent Vault. For a user who's in Agent Vault only through a group, name the group instead. + group_ids: With the "selected" scope, the groups whose members the machine identity can act on behalf of, up to 100. Each must be a member of Agent Vault. + + Returns: + The saved delegation, with missingAccessBundles. + """ + request_body = { + "scope": scope.value if isinstance(scope, AgentVaultDelegationScope) else scope, + "userIds": user_ids, + "groupIds": group_ids, + } + + result = self.requests.put( + path=f"/api/v1/agent-vault/delegations/{_segment(machine_identity_id)}", + json=request_body, + model=SingleAgentVaultDelegationResponse + ) + + return result.data.delegation + + def delete(self, machine_identity_id: str) -> AgentVaultDelegation: + """Stop a machine identity from creating Agent Vault sessions on behalf of users. The proxy refuses the sessions it already created on their behalf until a delegation covers those users again, so revoke the sessions to end them for good. + + Args: + machine_identity_id: The ID of the machine identity. + + Returns: + The deleted delegation. missingAccessBundles isn't set. + """ + # Not retried once the request may have reached the server: after a lost response, a retry would fail with a + # 404 for a delete that worked. + result = self.requests.delete_once( + path=f"/api/v1/agent-vault/delegations/{_segment(machine_identity_id)}", + model=SingleAgentVaultDelegationResponse + ) + + return result.data.delegation + + def add_subjects( + self, + machine_identity_id: str, + user_ids: Optional[List[str]] = None, + group_ids: Optional[List[str]] = None) -> AgentVaultDelegation: + """Add users and groups to a machine identity's delegation, keeping the ones it already names. If the machine identity has no delegation, this creates one with the "selected" scope. A delegation to all members can't take users or groups. + + Args: + machine_identity_id: The ID of the machine identity. + user_ids: The users to add, up to 100. Each must be a member of Agent Vault. Users the delegation already names are skipped. + group_ids: The groups to add, up to 100. Each must be a member of Agent Vault. Groups the delegation already names are skipped. + + Returns: + The delegation after the change, with missingAccessBundles. + """ + request_body = { + "userIds": user_ids, + "groupIds": group_ids, + } + + result = self.requests.post( + path=f"/api/v1/agent-vault/delegations/{_segment(machine_identity_id)}/subjects", + json=request_body, + model=SingleAgentVaultDelegationResponse + ) + + return result.data.delegation + + def remove_subjects( + self, + machine_identity_id: str, + user_ids: Optional[List[str]] = None, + group_ids: Optional[List[str]] = None) -> Optional[AgentVaultDelegation]: + """Take users and groups out of a machine identity's delegation. The proxy refuses the sessions it created on behalf of the users you take out until the delegation covers them again, so revoke the sessions to end them for good. Taking out the last user or group removes the delegation. + + Args: + machine_identity_id: The ID of the machine identity. + user_ids: The users to take out, up to 100. Users the delegation doesn't name are skipped. + group_ids: The groups to take out, up to 100. Groups the delegation doesn't name are skipped. + + Returns: + The delegation after the change, with missingAccessBundles, or None if it no longer names anyone. + """ + request_body = { + "userIds": user_ids, + "groupIds": group_ids, + } + + # Not retried once the request may have reached the server: after a lost response, a retry would fail with a + # 404 for a delete that worked. + result = self.requests.delete_once( + path=f"/api/v1/agent-vault/delegations/{_segment(machine_identity_id)}/subjects", + json=request_body, + model=SingleAgentVaultDelegationResponse + ) + + return result.data.delegation + + def list_available_subjects( + self, + machine_identity_id: str, + actor_type: Optional[Union[AgentVaultMemberType, str]] = None, + search: Optional[str] = None, + limit: int = 20, + offset: int = 0) -> ListAgentVaultMembersResponse: + """List the Agent Vault users and groups that can still be added to a machine identity's delegation. + + Args: + machine_identity_id: The ID of the machine identity. It must be a member of Agent Vault. + actor_type: List only users (AgentVaultMemberType.USER) or only groups (AgentVaultMemberType.GROUP). Defaults to both. + search: Match users by name, username or email address, and groups by name. + limit: The maximum number of users and groups to return (1-100). Defaults to 20. + offset: How many users and groups to skip. Defaults to 0. + + Returns: + ListAgentVaultMembersResponse containing the users and groups in members, and totalCount. + """ + params = { + "actorType": actor_type.value if isinstance(actor_type, AgentVaultMemberType) else actor_type, + "search": search, + "limit": limit, + "offset": offset, + } + + result = self.requests.get( + path=f"/api/v1/agent-vault/delegations/{_segment(machine_identity_id)}/subjects/available", + params=params, + model=ListAgentVaultMembersResponse + ) + + return result.data + + class AgentVault: """Manages Infisical Agent Vault.""" @@ -328,3 +877,5 @@ def __init__(self, requests: InfisicalRequests) -> None: self.requests = requests self.proxies = AgentVaultProxies(requests) self.sessions = AgentVaultSessions(requests) + self.access_bundles = AgentVaultAccessBundles(requests) + self.delegations = AgentVaultDelegations(requests) diff --git a/tests/conftest.py b/tests/conftest.py index 06020b1..9aa45c6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,9 +22,9 @@ def api(): @pytest.fixture def json_response(): - def make(payload): + def make(payload, status_code=200): response = requests.Response() - response.status_code = 200 + response.status_code = status_code response._content = json.dumps(payload).encode() return response return make diff --git a/tests/test_agent_vault.py b/tests/test_agent_vault.py index 2871a06..c8bc9f4 100644 --- a/tests/test_agent_vault.py +++ b/tests/test_agent_vault.py @@ -1,6 +1,14 @@ import pytest import requests +from infisical_sdk.api_types import ( + AgentVaultDelegationScope, + AgentVaultInlineService, + AgentVaultMemberType, + AgentVaultSessionActor, + AgentVaultTrafficPolicy, +) +from infisical_sdk.infisical_requests import APIError from infisical_sdk.resources.agent_vault import AgentVault PROXY = {"id": "a2b5e0d1-6f3e-4c55-9a43-3f0c1f0a2b7e", "name": "edge-proxy"} @@ -10,6 +18,58 @@ "status": "active", "createdAt": "2026-10-27T00:00:00.000Z", "accessBundle": {"id": "0d3f8a51-7c2e-4b9a-8e16-5a4c3b2d1f07", "name": "release"}, + "services": [], +} +ON_BEHALF_OF = { + "type": "user", + "id": "0d9a7c3e-5b1f-4e2a-8c6d-7f4b2e1a9c50", + "username": "jsmith@example.com", + "email": "jsmith@example.com", + "firstName": "Jane", + "lastName": "Smith", +} +BUNDLE = { + "id": "5e2d8c1a-7b3f-4a9e-8d6c-1f0b2a3c4d5e", + "name": "release", + "description": None, + "createdAt": "2026-10-27T00:00:00.000Z", + "updatedAt": "2026-10-27T00:00:00.000Z", +} +INLINE_SERVICE_INPUT = { + "type": "http", + "name": "github", + "hostPattern": "api.github.com", + "allowedMethods": ["GET"], + "credential": {"type": "bearer", "value": "github-token"}, + "customHeaders": [{"name": "X-Team", "value": "platform"}], +} +INLINE_SERVICE = { + "id": "8e2f4a6c-1b3d-4e5f-9a7b-2c4d6e8f0a1b", + "name": "github", + "type": "http", + "serverUrl": None, + "transport": None, + "hostPattern": "api.github.com", + "allowedMethods": ["GET"], + "allowedPathPrefixes": None, + "credential": {"type": "bearer", "headerName": "Authorization", "headerPrefix": "Bearer"}, + "customHeaders": [{"name": "X-Team", "prefix": ""}], + "substitutions": [], +} +MACHINE_IDENTITY_ID = "9c4b2a1d-3e5f-4a6b-8c7d-2e1f0a9b8c7d" +GROUP_ID = "7b1e4f2a-9c3d-4a6b-8e5f-1d2c3b4a5e6f" +SERVICE_ID = "3a5c7e9b-2d4f-4a6c-8e1b-5d7f9a2c4e6b" +DELEGATION = { + "machineIdentity": {"id": MACHINE_IDENTITY_ID, "name": "slack-bot"}, + "scope": "selected", + "users": [ + dict({key: ON_BEHALF_OF[key] for key in ("id", "username", "email", "firstName", "lastName")}, + addedAt="2026-10-27T00:00:00.000Z", missingAccessBundles=[{"id": BUNDLE["id"], "name": "release"}]), + ], + "groups": [{"id": GROUP_ID, "name": "Engineering", "addedAt": "2026-10-27T00:00:00.000Z", "missingAccessBundles": []}], + "createdAt": "2026-10-27T00:00:00.000Z", + "updatedAt": "2026-10-27T00:00:00.000Z", + "missingAccessBundles": [{"id": BUNDLE["id"], "name": "release"}], } @@ -18,9 +78,10 @@ def agent_vault(api): return AgentVault(api) -def test_proxies_update_sends_null_to_clear_allowed_hosts(api, agent_vault, json_response): +@pytest.mark.parametrize("allowed_hosts", [[], "", " "], ids=["empty list", "empty string", "blank string"]) +def test_proxies_update_sends_null_for_empty_allowed_hosts(api, agent_vault, json_response, allowed_hosts): api.session.patch.return_value = json_response({"proxy": PROXY}) - agent_vault.proxies.update(PROXY["id"], clear_allowed_hosts=True) + agent_vault.proxies.update(PROXY["id"], allowed_hosts=allowed_hosts) assert api.session.patch.call_args.kwargs["json"] == {"allowedHosts": None} @@ -30,10 +91,21 @@ def test_proxies_update_leaves_out_values_you_do_not_pass(api, agent_vault, json assert api.session.patch.call_args.kwargs["json"] == {"name": "edge-proxy"} -def test_proxies_update_rejects_allowed_hosts_with_clear_allowed_hosts(api, agent_vault): - with pytest.raises(ValueError): - agent_vault.proxies.update(PROXY["id"], allowed_hosts=["pypi.org"], clear_allowed_hosts=True) - api.session.patch.assert_not_called() +def test_proxies_create_sends_and_reads_the_description(api, agent_vault, json_response): + api.session.post.return_value = json_response({"proxy": dict(PROXY, description="Edge traffic"), "proxyToken": "token"}) + created = agent_vault.proxies.create(name="edge-proxy", description="Edge traffic") + assert api.session.post.call_args.kwargs["json"]["description"] == "Edge traffic" + assert created.proxy.description == "Edge traffic" + + +@pytest.mark.parametrize("kwargs, body", [ + ({"description": ""}, {"description": ""}), + ({"description": "", "allowed_hosts": []}, {"description": "", "allowedHosts": None}), +], ids=["description", "description and allowed hosts"]) +def test_proxies_update_sends_an_empty_description_to_remove_it(api, agent_vault, json_response, kwargs, body): + api.session.patch.return_value = json_response({"proxy": PROXY}) + agent_vault.proxies.update(PROXY["id"], **kwargs) + assert api.session.patch.call_args.kwargs["json"] == body @pytest.mark.parametrize("label", ["nightly deploy", None]) @@ -46,7 +118,8 @@ def test_sessions_update_always_sends_the_label(api, agent_vault, json_response, @pytest.mark.parametrize("create", [ lambda agent_vault: agent_vault.sessions.create(access_bundle="release"), lambda agent_vault: agent_vault.proxies.create(name="edge-proxy"), -], ids=["sessions.create", "proxies.create"]) + lambda agent_vault: agent_vault.access_bundles.create(name="release"), +], ids=["sessions.create", "proxies.create", "access_bundles.create"]) def test_creates_are_not_retried_once_the_request_may_have_reached_the_server(api, agent_vault, create): api.session.post.side_effect = requests.exceptions.ReadTimeout() with pytest.raises(requests.exceptions.ReadTimeout): @@ -54,18 +127,13 @@ def test_creates_are_not_retried_once_the_request_may_have_reached_the_server(ap assert api.session.post.call_count == 1 -def test_proxies_create_leaves_out_an_empty_allowed_hosts_list(api, agent_vault, json_response): +@pytest.mark.parametrize("allowed_hosts", [[], ""], ids=["empty list", "empty string"]) +def test_proxies_create_leaves_out_empty_allowed_hosts(api, agent_vault, json_response, allowed_hosts): api.session.post.return_value = json_response({"proxy": PROXY, "proxyToken": "token"}) - agent_vault.proxies.create(name="edge-proxy", allowed_hosts=[]) + agent_vault.proxies.create(name="edge-proxy", allowed_hosts=allowed_hosts) assert "allowedHosts" not in api.session.post.call_args.kwargs["json"] -def test_proxies_update_rejects_an_empty_allowed_hosts_list(api, agent_vault): - with pytest.raises(ValueError, match="clear_allowed_hosts"): - agent_vault.proxies.update(PROXY["id"], allowed_hosts=[]) - api.session.patch.assert_not_called() - - def test_proxies_rotate_token_is_not_retried_once_the_request_may_have_reached_the_server(api, agent_vault): api.session.post.side_effect = requests.exceptions.ReadTimeout() with pytest.raises(requests.exceptions.ReadTimeout): @@ -73,10 +141,11 @@ def test_proxies_rotate_token_is_not_retried_once_the_request_may_have_reached_t assert api.session.post.call_count == 1 -def test_sessions_revoke_leaves_the_access_bundle_unset_when_the_response_omits_it(api, agent_vault, json_response): - revoked = {key: value for key, value in SESSION.items() if key != "accessBundle"} +def test_sessions_revoke_leaves_the_access_bundle_and_services_unset_when_the_response_omits_them(api, agent_vault, json_response): + revoked = {key: value for key, value in SESSION.items() if key not in ("accessBundle", "services")} api.session.post.return_value = json_response({"session": dict(revoked, status="revoked")}) - assert agent_vault.sessions.revoke(SESSION["id"]).accessBundle is None + session = agent_vault.sessions.revoke(SESSION["id"]) + assert (session.accessBundle, session.services) == (None, None) def test_sessions_get_reads_the_access_bundle(api, agent_vault, json_response): @@ -97,3 +166,357 @@ def test_sessions_create_sends_one_access_bundle(api, agent_vault, json_response session = agent_vault.sessions.create(access_bundle="release", ttl="1h") assert api.session.post.call_args.kwargs["json"] == {"accessBundle": "release", "ttl": "1h"} assert session.accessBundle.name == "release" + + +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"] == {"accessBundle": "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"] == {"accessBundle": "release", "onBehalfOf": on_behalf_of} + + +def test_sessions_create_still_takes_the_access_bundle_first(api, agent_vault, json_response): + api.session.post.return_value = json_response({"session": dict(SESSION, token="token")}) + agent_vault.sessions.create("release", "1h") + assert api.session.post.call_args.kwargs["json"] == {"accessBundle": "release", "ttl": "1h"} + + +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 + + +def test_access_bundles_list_sends_booleans_the_way_the_api_reads_them(api, agent_vault, json_response): + api.session.get.return_value = json_response({"accessBundles": [], "totalCount": 0}) + agent_vault.access_bundles.list(requires_setup=True, granted=False) + params = api.session.get.call_args.kwargs["params"] + assert (params["requiresSetup"], params["granted"]) == ("true", "false") + + +def test_access_bundles_list_names_the_user_a_session_would_act_for(api, agent_vault, json_response): + api.session.get.return_value = json_response({"accessBundles": [], "totalCount": 0}) + agent_vault.access_bundles.list(on_behalf_of_user_email=ON_BEHALF_OF["email"]) + params = api.session.get.call_args.kwargs["params"] + assert (params["onBehalfOfUserEmail"], params["onBehalfOfUserId"]) == (ON_BEHALF_OF["email"], None) + + +def test_access_bundles_list_rejects_both_a_user_email_and_a_user_id(api, agent_vault): + with pytest.raises(ValueError, match="not both"): + agent_vault.access_bundles.list(on_behalf_of_user_email=ON_BEHALF_OF["email"], on_behalf_of_user_id=ON_BEHALF_OF["id"]) + api.session.get.assert_not_called() + + +def test_access_bundles_list_reads_the_services_requiring_setup(api, agent_vault, json_response): + listed = dict(BUNDLE, serviceCount=1, memberCount=2, hostPatterns=["api.github.com"], + servicesRequiringSetup=[{"id": SERVICE_ID, "name": "github"}]) + api.session.get.return_value = json_response({"accessBundles": [listed], "totalCount": 1}) + bundle = agent_vault.access_bundles.list().accessBundles[0] + assert [(service.id, service.name) for service in bundle.servicesRequiringSetup] == [(SERVICE_ID, "github")] + assert bundle.services is None + + +def test_access_bundles_get_reads_services_and_keeps_their_nested_settings(api, agent_vault, json_response): + service = { + "id": SERVICE_ID, "accessBundleId": BUNDLE["id"], "name": "github", "type": "http", + "hostPattern": "api.github.com", "credential": {"type": "bearer", "memberProvidedFields": []}, + "customHeaders": [], "substitutions": [], "personalValues": None, + "createdAt": BUNDLE["createdAt"], "updatedAt": BUNDLE["updatedAt"], + } + api.session.get.return_value = json_response({"accessBundle": dict(BUNDLE, services=[service])}) + [read] = agent_vault.access_bundles.get(BUNDLE["id"]).services + assert (read.name, read.credential["type"], read.variableReferences) == ("github", "bearer", None) + + +def test_access_bundles_update_sends_an_empty_description_to_remove_it(api, agent_vault, json_response): + api.session.patch.return_value = json_response({"accessBundle": BUNDLE}) + agent_vault.access_bundles.update(BUNDLE["id"], description="") + assert api.session.patch.call_args.kwargs["json"] == {"description": ""} + + +def test_access_bundles_grant_and_revoke_send_only_the_grantees_you_name(api, agent_vault, json_response): + change = {"members": [{"type": "user", "id": ON_BEHALF_OF["id"]}], "skipped": []} + api.session.post.return_value = json_response(change) + api.session.delete.return_value = json_response(change) + agent_vault.access_bundles.grant(BUNDLE["id"], machine_identity_ids=[MACHINE_IDENTITY_ID]) + revoked = agent_vault.access_bundles.revoke(BUNDLE["id"], user_ids=[ON_BEHALF_OF["id"]]) + assert api.session.post.call_args.kwargs["json"] == {"machineIdentityIds": [MACHINE_IDENTITY_ID]} + assert api.session.delete.call_args.kwargs["json"] == {"userIds": [ON_BEHALF_OF["id"]]} + assert [(member.type, member.id) for member in revoked.members] == [("user", ON_BEHALF_OF["id"])] + + +def test_delegations_set_puts_the_scope_value(api, agent_vault, json_response): + api.session.put.return_value = json_response({"delegation": DELEGATION}) + agent_vault.delegations.set(MACHINE_IDENTITY_ID, AgentVaultDelegationScope.SELECTED, group_ids=[GROUP_ID]) + assert api.session.put.call_args.kwargs["json"] == {"scope": "selected", "groupIds": [GROUP_ID]} + + +def test_delegations_get_reads_users_groups_and_missing_access_bundles(api, agent_vault, json_response): + api.session.get.return_value = json_response({"delegation": DELEGATION}) + delegation = agent_vault.delegations.get(MACHINE_IDENTITY_ID) + assert delegation.machineIdentity.name == "slack-bot" + assert [bundle.name for bundle in delegation.users[0].missingAccessBundles] == ["release"] + assert (delegation.groups[0].name, delegation.groups[0].missingAccessBundles) == ("Engineering", []) + + +def test_delegations_list_joins_the_machine_identity_ids(api, agent_vault, json_response): + other_machine_identity_id = "2f6e8a4c-1b3d-4c5e-9f7a-6b8c0d2e4f1a" + listed = {key: value for key, value in DELEGATION.items() if key != "missingAccessBundles"} + api.session.get.return_value = json_response({"delegations": [listed], "totalCount": 1}) + delegations = agent_vault.delegations.list(machine_identity_ids=[MACHINE_IDENTITY_ID, other_machine_identity_id]) + expected = f"{MACHINE_IDENTITY_ID},{other_machine_identity_id}" + assert api.session.get.call_args.kwargs["params"]["machineIdentityIds"] == expected + assert delegations.delegations[0].missingAccessBundles is None + + +def test_delegations_remove_subjects_returns_none_once_the_last_one_is_gone(api, agent_vault, json_response): + api.session.delete.return_value = json_response({"delegation": None}) + assert agent_vault.delegations.remove_subjects(MACHINE_IDENTITY_ID, user_ids=[ON_BEHALF_OF["id"]]) is None + assert api.session.delete.call_args.kwargs["json"] == {"userIds": [ON_BEHALF_OF["id"]]} + + +def test_delegations_list_available_subjects_sends_the_actor_type_value(api, agent_vault, json_response): + api.session.get.return_value = json_response({"members": [], "totalCount": 0}) + agent_vault.delegations.list_available_subjects(MACHINE_IDENTITY_ID, actor_type=AgentVaultMemberType.GROUP) + assert api.session.get.call_args.kwargs["params"]["actorType"] == "group" + + +@pytest.mark.parametrize("traffic_policy", [ + AgentVaultTrafficPolicy.SERVICE_HOSTS, AgentVaultTrafficPolicy.BUNDLE_HOSTS, "service-hosts", "bundle-hosts", +], ids=["SERVICE_HOSTS", "BUNDLE_HOSTS", "service-hosts", "bundle-hosts"]) +def test_proxies_send_either_name_of_service_hosts_as_bundle_hosts_which_every_version_takes( + api, agent_vault, json_response, traffic_policy): + api.session.post.return_value = json_response({"proxy": PROXY, "proxyToken": "token"}) + api.session.patch.return_value = json_response({"proxy": PROXY}) + agent_vault.proxies.create(name="edge-proxy", traffic_policy=traffic_policy) + agent_vault.proxies.update(PROXY["id"], traffic_policy=traffic_policy) + assert api.session.post.call_args.kwargs["json"]["trafficPolicy"] == "bundle-hosts" + assert api.session.patch.call_args.kwargs["json"] == {"trafficPolicy": "bundle-hosts"} + + +@pytest.mark.parametrize("kwargs", [{}, {"traffic_policy": AgentVaultTrafficPolicy.ANY_HOST}, {"traffic_policy": "any-host"}], + ids=["default", "ANY_HOST", "any-host"]) +def test_proxies_create_sends_any_host_as_it_is(api, agent_vault, json_response, kwargs): + api.session.post.return_value = json_response({"proxy": PROXY, "proxyToken": "token"}) + agent_vault.proxies.create(name="edge-proxy", **kwargs) + assert api.session.post.call_args.kwargs["json"]["trafficPolicy"] == "any-host" + + +@pytest.mark.parametrize("reported", ["service-hosts", "bundle-hosts"], ids=["current version", "version before the rename"]) +@pytest.mark.parametrize("verb, payload, call", [ + ("get", lambda proxy: {"proxies": [proxy], "totalCount": 1}, lambda agent_vault: agent_vault.proxies.list().proxies[0]), + ("post", lambda proxy: {"proxy": proxy, "proxyToken": "token"}, + lambda agent_vault: agent_vault.proxies.create(name="edge-proxy").proxy), + ("patch", lambda proxy: {"proxy": proxy}, lambda agent_vault: agent_vault.proxies.update(PROXY["id"], poll_interval=30)), + ("post", lambda proxy: {"proxy": proxy, "proxyToken": "token"}, + lambda agent_vault: agent_vault.proxies.rotate_token(PROXY["id"]).proxy), + ("post", lambda proxy: {"proxy": proxy}, lambda agent_vault: agent_vault.proxies.revoke(PROXY["id"])), +], ids=["list", "create", "update", "rotate_token", "revoke"]) +def test_proxies_report_either_name_as_service_hosts(api, agent_vault, json_response, reported, verb, payload, call): + getattr(api.session, verb).return_value = json_response(payload(dict(PROXY, trafficPolicy=reported))) + proxy = call(agent_vault) + assert proxy.trafficPolicy == "service-hosts" + assert proxy.trafficPolicy == AgentVaultTrafficPolicy.SERVICE_HOSTS + assert proxy.trafficPolicy == AgentVaultTrafficPolicy.BUNDLE_HOSTS + + +def test_proxies_list_leaves_the_traffic_policy_unset_for_a_member(api, agent_vault, json_response): + api.session.get.return_value = json_response({"proxies": [PROXY], "totalCount": 1}) + assert agent_vault.proxies.list().proxies[0].trafficPolicy is None + + +def test_traffic_policy_old_name_is_the_same_member(): + assert AgentVaultTrafficPolicy.BUNDLE_HOSTS is AgentVaultTrafficPolicy.SERVICE_HOSTS + assert AgentVaultTrafficPolicy("bundle-hosts") is AgentVaultTrafficPolicy.SERVICE_HOSTS + assert AgentVaultTrafficPolicy("service-hosts") is AgentVaultTrafficPolicy.SERVICE_HOSTS + assert list(AgentVaultTrafficPolicy) == [AgentVaultTrafficPolicy.ANY_HOST, AgentVaultTrafficPolicy.SERVICE_HOSTS] + with pytest.raises(ValueError): + AgentVaultTrafficPolicy("all-hosts") + + +def test_sessions_create_sends_services_without_an_access_bundle(api, agent_vault, json_response): + api.session.post.return_value = json_response({"session": dict(SESSION, token="token", services=[INLINE_SERVICE])}) + agent_vault.sessions.create(services=[INLINE_SERVICE_INPUT]) + assert api.session.post.call_args.kwargs["json"] == {"services": [INLINE_SERVICE_INPUT]} + + +def test_sessions_create_sends_an_access_bundle_and_services_together(api, agent_vault, json_response): + api.session.post.return_value = json_response({"session": dict(SESSION, token="token", services=[INLINE_SERVICE])}) + agent_vault.sessions.create(access_bundle="release", services=[INLINE_SERVICE_INPUT]) + assert api.session.post.call_args.kwargs["json"] == {"accessBundle": "release", "services": [INLINE_SERVICE_INPUT]} + + +def test_sessions_create_sends_an_empty_access_bundle_name_so_the_server_rejects_it(api, agent_vault, json_response): + api.session.post.return_value = json_response({"session": dict(SESSION, token="token")}) + agent_vault.sessions.create(access_bundle="") + assert api.session.post.call_args.kwargs["json"] == {"accessBundle": ""} + + +def test_sessions_read_services_as_the_api_sends_them(api, agent_vault, json_response): + session = dict(SESSION, services=[INLINE_SERVICE]) + api.session.post.return_value = json_response({"session": dict(session, token="token")}) + api.session.get.side_effect = [ + json_response({"session": session}), + json_response({"sessions": [session], "totalCount": 1}), + ] + expected = [AgentVaultInlineService(**INLINE_SERVICE)] + assert agent_vault.sessions.create(services=[INLINE_SERVICE_INPUT]).services == expected + assert agent_vault.sessions.get(SESSION["id"]).services == expected + assert agent_vault.sessions.list().sessions[0].services == expected + + +def test_sessions_from_an_infisical_version_without_inline_services_have_an_empty_list_of_them( + api, agent_vault, json_response): + older = {key: value for key, value in SESSION.items() if key != "services"} + api.session.post.return_value = json_response({"session": dict(older, token="token")}) + api.session.patch.return_value = json_response({"session": older}) + api.session.get.side_effect = [ + json_response({"session": older}), + json_response({"sessions": [older], "totalCount": 1}), + ] + assert agent_vault.sessions.create(access_bundle="release").services == [] + assert agent_vault.sessions.update(SESSION["id"], None).services == [] + assert agent_vault.sessions.get(SESSION["id"]).services == [] + assert agent_vault.sessions.list().sessions[0].services == [] + + +@pytest.mark.parametrize("status_code, payload, explanation", [ + (422, { + "reqId": "req-1", + "statusCode": 422, + "message": [{ + "code": "too_big", + "maximum": 20, + "path": ["services"], + "message": "A session can have at most 20 services. Add the others to an access bundle and name it in accessBundles.", + }], + "error": "ValidationFailure", + }, "at most 20 services"), + (403, { + "statusCode": 403, + "message": "Only an Agent Vault admin can send services with a session. Name an access bundle instead.", + "error": "PermissionDenied", + }, "Only an Agent Vault admin can send services with a session"), +], ids=["validation error", "refusal"]) +def test_sessions_create_keeps_the_servers_explanation_when_it_refuses_services( + api, agent_vault, json_response, status_code, payload, explanation): + api.session.post.return_value = json_response(payload, status_code) + with pytest.raises(APIError) as raised: + agent_vault.sessions.create(services=[INLINE_SERVICE_INPUT]) + assert raised.value.status_code == status_code + assert explanation in str(raised.value) + assert api.session.post.call_count == 1 + + +def test_access_bundles_get_reads_the_tools_an_mcp_service_allows(api, agent_vault, json_response): + service = { + "id": SERVICE_ID, "accessBundleId": BUNDLE["id"], "name": "linear", "type": "mcp", + "serverUrl": "https://mcp.linear.app/mcp", "transport": "streamable-http", "hostPattern": "mcp.linear.app", + "allowedMethods": None, "allowedPathPrefixes": None, "allowedTools": ["list_issues"], + "credential": {"type": "bearer", "memberProvidedFields": []}, "customHeaders": [], "substitutions": [], + "createdAt": BUNDLE["createdAt"], "updatedAt": BUNDLE["updatedAt"], + } + api.session.get.return_value = json_response({"accessBundle": dict(BUNDLE, services=[service])}) + assert agent_vault.access_bundles.get(BUNDLE["id"]).services[0].allowedTools == ["list_issues"] + + +def test_access_bundles_list_grantees_reads_a_pending_organization_membership(api, agent_vault, json_response): + grantee = dict({key: ON_BEHALF_OF[key] for key in ("type", "id", "username", "email", "firstName", "lastName")}, + isOrgMembershipPending=True, grantedAt="2026-10-27T00:00:00.000Z") + api.session.get.return_value = json_response({"members": [grantee], "totalCount": 1}) + assert agent_vault.access_bundles.list_grantees(BUNDLE["id"]).members[0].isOrgMembershipPending is True + + +@pytest.mark.parametrize("bad_id", ["", ".", "..", None], ids=["empty", "dot", "dot dot", "None"]) +def test_ids_that_would_reach_another_route_are_refused_before_any_request(api, agent_vault, bad_id): + # The URL would resolve ".." like a directory, so revoke("..") would reach the route that removes members from + # Agent Vault, which takes the same body. + with pytest.raises(ValueError, match="Invalid ID"): + agent_vault.access_bundles.revoke(bad_id, user_ids=[ON_BEHALF_OF["id"]]) + with pytest.raises(ValueError, match="Invalid ID"): + agent_vault.sessions.get(bad_id) + api.session.delete.assert_not_called() + api.session.get.assert_not_called() + + +@pytest.mark.parametrize("raw_id, segment", [ + ("a/../..", "a%2F..%2F.."), + ("%2e%2e", "%252e%252e"), +], ids=["slashes", "encoded dots"]) +def test_ids_are_sent_as_one_path_segment(api, agent_vault, json_response, raw_id, segment): + api.session.get.return_value = json_response({"session": SESSION}) + agent_vault.sessions.get(raw_id) + url = api.session.get.call_args.args[0] + # The path that requests sends, after it normalizes the URL. + assert requests.Request("GET", url).prepare().path_url == f"/api/v1/agent-vault/sessions/{segment}" + + +@pytest.mark.parametrize("verb, call", [ + ("post", lambda agent_vault: agent_vault.access_bundles.grant(BUNDLE["id"], user_ids=[ON_BEHALF_OF["id"]])), + ("delete", lambda agent_vault: agent_vault.access_bundles.revoke(BUNDLE["id"], user_ids=[ON_BEHALF_OF["id"]])), +], ids=["grant", "revoke"]) +def test_grant_and_revoke_are_not_retried_once_the_request_may_have_reached_the_server(api, agent_vault, verb, call): + # After a lost response, a retry would report each grantee as skipped. + getattr(api.session, verb).side_effect = requests.exceptions.ReadTimeout() + with pytest.raises(requests.exceptions.ReadTimeout): + call(agent_vault) + assert getattr(api.session, verb).call_count == 1 + + +def test_revoke_is_retried_when_the_connection_never_opened(api, agent_vault, json_response): + change = {"members": [{"type": "user", "id": ON_BEHALF_OF["id"]}], "skipped": []} + api.session.delete.side_effect = [requests.exceptions.ConnectTimeout(), json_response(change)] + revoked = agent_vault.access_bundles.revoke(BUNDLE["id"], user_ids=[ON_BEHALF_OF["id"]]) + assert [member.id for member in revoked.members] == [ON_BEHALF_OF["id"]] + assert api.session.delete.call_count == 2 + + +def test_access_bundles_grantee_lists_send_the_actor_type_value(api, agent_vault, json_response): + api.session.get.return_value = json_response({"members": [], "totalCount": 0}) + agent_vault.access_bundles.list_grantees(BUNDLE["id"], actor_type=AgentVaultMemberType.GROUP) + assert api.session.get.call_args.kwargs["params"]["actorType"] == "group" + agent_vault.access_bundles.list_available_grantees(BUNDLE["id"], actor_type="machineIdentity") + assert api.session.get.call_args.kwargs["params"]["actorType"] == "machineIdentity" + + +def test_api_error_is_exported_from_the_package_root(): + import infisical_sdk + from infisical_sdk.infisical_requests import APIError as RequestsAPIError + assert infisical_sdk.APIError is RequestsAPIError + assert issubclass(infisical_sdk.APIError, infisical_sdk.InfisicalError) + + +@pytest.mark.parametrize("call", [ + lambda agent_vault: agent_vault.proxies.delete(PROXY["id"]), + lambda agent_vault: agent_vault.access_bundles.delete(BUNDLE["id"]), + lambda agent_vault: agent_vault.delegations.delete(MACHINE_IDENTITY_ID), + lambda agent_vault: agent_vault.delegations.remove_subjects(MACHINE_IDENTITY_ID, user_ids=[ON_BEHALF_OF["id"]]), +], ids=["proxies.delete", "access_bundles.delete", "delegations.delete", "delegations.remove_subjects"]) +def test_deletes_are_not_retried_once_the_request_may_have_reached_the_server(api, agent_vault, call): + # After a lost response, a retry would fail with a 404 for a delete that worked. + api.session.delete.side_effect = requests.exceptions.ReadTimeout() + with pytest.raises(requests.exceptions.ReadTimeout): + call(agent_vault) + assert api.session.delete.call_count == 1 diff --git a/tests/test_infisical_requests.py b/tests/test_infisical_requests.py index 248679a..dfa747e 100644 --- a/tests/test_infisical_requests.py +++ b/tests/test_infisical_requests.py @@ -20,6 +20,18 @@ def test_patch_keeps_false_zero_and_empty_string(api, json_response): assert api.session.patch.call_args.kwargs["json"] == {"enabled": False, "count": 0, "note": ""} +def test_put_leaves_out_none_values(api, json_response): + api.session.put.return_value = json_response({}) + api.put(path="/api/v1/example", model=dict, json={"scope": "selected", "groupIds": None}) + assert api.session.put.call_args.kwargs["json"] == {"scope": "selected"} + + +def test_put_retries_a_read_timeout(api, json_response): + api.session.put.side_effect = [requests.exceptions.ReadTimeout(), json_response({"ok": True})] + assert api.put(path="/api/v1/example", model=dict).data == {"ok": True} + assert api.session.put.call_count == 2 + + def test_post_retries_a_read_timeout(api, json_response): api.session.post.side_effect = [requests.exceptions.ReadTimeout(), json_response({"ok": True})] assert api.post(path="/api/v1/example", model=dict).data == {"ok": True}