feat(agent-vault): sessions on behalf of users, access bundles, delegations and proxy descriptions

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
+from .infisical_requests import APIError, InfisicalError # noqa
+from .api_types import SingleSecretResponse, ListSecretsResponse, BaseSecret, SymmetricEncryption, DynamicSecretProviders, AgentVaultTrafficPolicy, AgentVaultSessionStatus, AgentVaultSessionScope, AgentVaultMemberType, AgentVaultDelegationScope # noqa

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'],
+        )