feat(agent-vault): add Agent Vault proxies and sessions

.github/workflows/python.yml

@@ -34,7 +34,7 @@ jobs:
           # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
           flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
 
-      # ? Currently no tests are available
-      #- name: Test with pytest
-      #  run: |
-      #    pytest
+      # Only tests/: the scripts in sink/ call a live Infisical instance
+      - name: Test with pytest
+        run: |
+          python -m pytest tests

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

infisical_sdk/api_types.py

@@ -448,3 +448,168 @@ def from_dict(cls, data: Dict) -> 'SingleLeaseResponse':
         return cls(
             lease=DynamicSecretLease.from_dict(data['lease']),
         )
+
+class AgentVaultTrafficPolicy(str, Enum):
+    """Enum for which hosts an agent may reach through an Agent Vault proxy"""
+    ANY_HOST = "any-host"
+    BUNDLE_HOSTS = "bundle-hosts"
+
+class AgentVaultSessionStatus(str, Enum):
+    """Enum for Agent Vault session status"""
+    ACTIVE = "active"
+    REVOKED = "revoked"
+    EXPIRED = "expired"
+
+class AgentVaultSessionScope(str, Enum):
+    """Enum for whose Agent Vault sessions to list"""
+    MINE = "mine"
+    ALL = "all"
+
+@dataclass
+class AgentVaultProxy(BaseModel):
+    """Infisical Agent Vault Proxy. Only admins see the settings fields, so they are None for members."""
+    id: str
+    name: str
+    heartbeat: Optional[str] = None
+    isHealthy: Optional[bool] = None
+    rootCaFingerprint: Optional[str] = None
+    rootCaExpiresAt: Optional[str] = None
+    trafficPolicy: Optional[str] = None
+    allowedHosts: Optional[str] = None
+    pollInterval: Optional[int] = None
+    createdAt: Optional[str] = None
+
+@dataclass
+class SingleAgentVaultProxyResponse(BaseModel):
+    """Response model for update/delete/revoke Agent Vault proxy API"""
+    proxy: AgentVaultProxy
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'SingleAgentVaultProxyResponse':
+        return cls(
+            proxy=AgentVaultProxy.from_dict(data['proxy']),
+        )
+
+@dataclass
+class AgentVaultProxyWithTokenResponse(BaseModel):
+    """Response model for create Agent Vault proxy and rotate proxy token API. The token is only returned once."""
+    proxy: AgentVaultProxy
+    proxyToken: str
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'AgentVaultProxyWithTokenResponse':
+        return cls(
+            proxy=AgentVaultProxy.from_dict(data['proxy']),
+            proxyToken=data['proxyToken'],
+        )
+
+@dataclass
+class ListAgentVaultProxiesResponse(BaseModel):
+    """Response model for list Agent Vault proxies API"""
+    proxies: List[AgentVaultProxy]
+    totalCount: int
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'ListAgentVaultProxiesResponse':
+        return cls(
+            proxies=[AgentVaultProxy.from_dict(proxy) for proxy in data['proxies']],
+            totalCount=data['totalCount'],
+        )
+
+@dataclass
+class AgentVaultSessionActor(BaseModel):
+    """The user or machine identity an Agent Vault session belongs to"""
+    type: str
+    id: Optional[str] = None
+    username: Optional[str] = None
+    email: Optional[str] = None
+    firstName: Optional[str] = None
+    lastName: Optional[str] = None
+    name: Optional[str] = None
+
+@dataclass
+class AgentVaultSessionAccessBundle(BaseModel):
+    """An access bundle carried by an Agent Vault session"""
+    name: str
+    position: int
+    id: Optional[str] = None
+
+@dataclass
+class AgentVaultSession(BaseModel):
+    """Infisical Agent Vault Session"""
+    id: str
+    status: str
+    createdAt: str
+    label: Optional[str] = None
+    actor: Optional[AgentVaultSessionActor] = None
+    expiresAt: Optional[str] = None
+    revokedAt: Optional[str] = None
+    accessBundles: Optional[List[AgentVaultSessionAccessBundle]] = None
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'AgentVaultSession':
+        """Create model from dictionary with nested actor and access bundles. accessBundles is None if the response leaves it out."""
+        session_data = data.copy()
+        if data.get('actor') is not None:
+            session_data['actor'] = AgentVaultSessionActor.from_dict(data['actor'])
+        if data.get('accessBundles') is not None:
+            session_data['accessBundles'] = [
+                AgentVaultSessionAccessBundle.from_dict(bundle) for bundle in data['accessBundles']
+            ]
+
+        return super().from_dict(session_data)
+
+@dataclass
+class CreatedAgentVaultSession(BaseModel):
+    """Infisical Agent Vault Session as returned at creation. The token is only returned once."""
+    id: str
+    token: str
+    createdAt: str
+    label: Optional[str] = None
+    expiresAt: Optional[str] = None
+    accessBundles: List[AgentVaultSessionAccessBundle] = field(default_factory=list)
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'CreatedAgentVaultSession':
+        """Create model from dictionary with nested access bundles"""
+        session_data = data.copy()
+        session_data['accessBundles'] = [
+            AgentVaultSessionAccessBundle.from_dict(bundle) for bundle in data.get('accessBundles', [])
+        ]
+
+        return super().from_dict(session_data)
+
+@dataclass
+class SingleAgentVaultSessionResponse(BaseModel):
+    """Response model for get/revoke Agent Vault session API"""
+    session: AgentVaultSession
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'SingleAgentVaultSessionResponse':
+        return cls(
+            session=AgentVaultSession.from_dict(data['session']),
+        )
+
+@dataclass
+class CreateAgentVaultSessionResponse(BaseModel):
+    """Response model for create Agent Vault session API"""
+    session: CreatedAgentVaultSession
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'CreateAgentVaultSessionResponse':
+        return cls(
+            session=CreatedAgentVaultSession.from_dict(data['session']),
+        )
+
+@dataclass
+class ListAgentVaultSessionsResponse(BaseModel):
+    """Response model for list Agent Vault sessions API"""
+    sessions: List[AgentVaultSession]
+    totalCount: int
+
+    @classmethod
+    def from_dict(cls, data: Dict) -> 'ListAgentVaultSessionsResponse':
+        return cls(
+            sessions=[AgentVaultSession.from_dict(session) for session in data['sessions']],
+            totalCount=data['totalCount'],
+        )

infisical_sdk/client.py

@@ -5,6 +5,7 @@
 from infisical_sdk.resources import KMS
 from infisical_sdk.resources import V2Folders
 from infisical_sdk.resources import DynamicSecrets
+from infisical_sdk.resources import AgentVault
 
 from infisical_sdk.util import SecretsCache
 
@@ -28,6 +29,7 @@ def __init__(self, host: str, token: str = None, cache_ttl: int = 60):
         self.kms = KMS(self.api)
         self.folders = V2Folders(self.api)
         self.dynamic_secrets = DynamicSecrets(self.api)
+        self.agent_vault = AgentVault(self.api)
 
     def set_token(self, token: str):
         """

infisical_sdk/infisical_requests.py

@@ -22,6 +22,11 @@
     ConnectionAbortedError,
 ]
 
+# Raised before a request reaches the server, so retrying can't make the server handle it twice
+CONNECT_ERRORS = [
+    requests.exceptions.ConnectTimeout,
+]
+
 def join_url(base: str, path: str) -> str:
     """
     Join base URL and path properly, handling slashes appropriately.
@@ -199,20 +204,47 @@ def post(
             headers=dict(response.headers)
         )
 
-    @with_retry(max_retries=4, base_delay=1.0)
-    def patch(
+    @with_retry(max_retries=4, base_delay=1.0, network_errors=CONNECT_ERRORS)
+    def post_once(
             self,
             path: str,
             model: Type[T],
             json: Optional[Dict[str, Any]] = None
           ) -> APIResponse[T]:
 
-        """Make a PATCH request with JSON data"""
+        """Make a POST request that must not run twice, such as one that creates something.
+        It's only retried if the connection never opened, so a lost response can't run it again."""
 
         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.post(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 patch(
+            self,
+            path: str,
+            model: Type[T],
+            json: Optional[Dict[str, Any]] = None,
+            nullable_keys: Optional[List[str]] = None
+          ) -> APIResponse[T]:
+
+        """Make a PATCH request with JSON data. Keys in nullable_keys are sent even when None, to clear a value."""
+
+        if json is not None:
+            # Filter out None values
+            json = {k: v for k, v in json.items() if v is not None or k in (nullable_keys or [])}
+
         response = self.session.patch(self._build_url(path), json=json)
         data = self._handle_response(response)
 

infisical_sdk/resources/__init__.py

@@ -2,4 +2,5 @@
 from .kms import KMS
 from .auth import Auth
 from .folders import V2Folders
-from .dynamic_secrets import DynamicSecrets
+from .dynamic_secrets import DynamicSecrets
+from .agent_vault import AgentVault