# """Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.""" # from .basesdk import BaseSDK # from typing import Any, Mapping, Optional, Union, cast # from unstructured_client import utils # from unstructured_client._hooks import HookContext # from unstructured_client.models import errors, operations, shared # from unstructured_client.types import BaseModel, OptionalNullable, UNSET # from unstructured_client.utils.unmarshal_json_response import unmarshal_json_response # # region imports # from cryptography import x509 # from cryptography.hazmat.primitives import serialization, hashes # from cryptography.hazmat.primitives.asymmetric import padding, rsa # from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes # from cryptography.hazmat.backends import default_backend # import os # import base64 # # endregion imports # class Users(BaseSDK): # # region sdk-class-body # def _encrypt_rsa_aes( # self, # public_key: rsa.RSAPublicKey, # plaintext: str, # ) -> dict: # # Generate a random AES key # aes_key = os.urandom(32) # 256-bit AES key # # Generate a random IV # iv = os.urandom(16) # # Encrypt using AES-CFB # cipher = Cipher( # algorithms.AES(aes_key), # modes.CFB(iv), # ) # encryptor = cipher.encryptor() # ciphertext = encryptor.update(plaintext.encode("utf-8")) + encryptor.finalize() # # Encrypt the AES key using the RSA public key # encrypted_key = public_key.encrypt( # aes_key, # padding.OAEP( # mgf=padding.MGF1(algorithm=hashes.SHA256()), # algorithm=hashes.SHA256(), # label=None, # ), # ) # return { # "encrypted_aes_key": base64.b64encode(encrypted_key).decode("utf-8"), # "aes_iv": base64.b64encode(iv).decode("utf-8"), # "encrypted_value": base64.b64encode(ciphertext).decode("utf-8"), # "type": "rsa_aes", # } # def _encrypt_rsa( # self, # public_key: rsa.RSAPublicKey, # plaintext: str, # ) -> dict: # # Load public RSA key # ciphertext = public_key.encrypt( # plaintext.encode(), # padding.OAEP( # mgf=padding.MGF1(algorithm=hashes.SHA256()), # algorithm=hashes.SHA256(), # label=None, # ), # ) # return { # "encrypted_value": base64.b64encode(ciphertext).decode("utf-8"), # "type": "rsa", # "encrypted_aes_key": "", # "aes_iv": "", # } # def decrypt_secret( # self, # private_key_pem: str, # encrypted_value: str, # secret_type: str, # encrypted_aes_key: str, # aes_iv: str, # ) -> str: # private_key = serialization.load_pem_private_key( # private_key_pem.encode("utf-8"), password=None, backend=default_backend() # ) # if not isinstance(private_key, rsa.RSAPrivateKey): # raise TypeError("Private key must be a RSA private key for decryption.") # if secret_type == "rsa": # ciphertext = base64.b64decode(encrypted_value) # plaintext = private_key.decrypt( # ciphertext, # padding.OAEP( # mgf=padding.MGF1(algorithm=hashes.SHA256()), # algorithm=hashes.SHA256(), # label=None, # ), # ) # return plaintext.decode("utf-8") # # aes_rsa # encrypted_aes_key_decoded = base64.b64decode(encrypted_aes_key) # iv = base64.b64decode(aes_iv) # ciphertext = base64.b64decode(encrypted_value) # aes_key = private_key.decrypt( # encrypted_aes_key_decoded, # padding.OAEP( # mgf=padding.MGF1(algorithm=hashes.SHA256()), # algorithm=hashes.SHA256(), # label=None, # ), # ) # cipher = Cipher( # algorithms.AES(aes_key), # modes.CFB(iv), # ) # decryptor = cipher.decryptor() # plaintext = decryptor.update(ciphertext) + decryptor.finalize() # return plaintext.decode("utf-8") # def encrypt_secret( # self, # encryption_cert_or_key_pem: str, # plaintext: str, # encryption_type: Optional[str] = None, # ) -> dict: # """ # Encrypts a plaintext string for securely sending to the Unstructured API. # Args: # encryption_cert_or_key_pem (str): A PEM-encoded RSA public key or certificate. # plaintext (str): The string to encrypt. # type (str, optional): Encryption type, either "rsa" or "rsa_aes". # Returns: # dict: A dictionary with encrypted AES key, iv, and ciphertext (all base64-encoded). # """ # # If a cert is provided, extract the public key # if "BEGIN CERTIFICATE" in encryption_cert_or_key_pem: # cert = x509.load_pem_x509_certificate( # encryption_cert_or_key_pem.encode("utf-8"), # ) # public_key = cert.public_key() # type: ignore[assignment] # else: # public_key = serialization.load_pem_public_key( # encryption_cert_or_key_pem.encode("utf-8"), backend=default_backend() # ) # type: ignore[assignment] # if not isinstance(public_key, rsa.RSAPublicKey): # raise TypeError("Public key must be a RSA public key for encryption.") # # If the plaintext is short, use RSA directly # # Otherwise, use a RSA_AES envelope hybrid # # Use the length of the public key to determine the encryption type # key_size_bytes = public_key.key_size // 8 # max_rsa_length = key_size_bytes - 66 # OAEP SHA256 overhead # if not encryption_type: # encryption_type = "rsa" if len(plaintext) <= max_rsa_length else "rsa_aes" # if encryption_type == "rsa": # return self._encrypt_rsa(public_key, plaintext) # return self._encrypt_rsa_aes(public_key, plaintext) # # endregion sdk-class-body # def get_encryption_certificate( # self, # *, # request: Union[ # operations.GetEncryptionCertificateRequest, # operations.GetEncryptionCertificateRequestTypedDict, # ], # retries: OptionalNullable[utils.RetryConfig] = UNSET, # server_url: Optional[str] = None, # timeout_ms: Optional[int] = None, # http_headers: Optional[Mapping[str, str]] = None, # ) -> operations.GetEncryptionCertificateResponse: # r"""Retrieve the user's public key for encryption. # Retrieve a short lived certificate with the public key for encrypting secrets. # :param request: The request object to send. # :param retries: Override the default retry configuration for this method # :param server_url: Override the default server URL for this method # :param timeout_ms: Override the default request timeout configuration for this method in milliseconds # :param http_headers: Additional headers to set or replace on requests. # """ # base_url = None # url_variables = None # if timeout_ms is None: # timeout_ms = self.sdk_configuration.timeout_ms # if server_url is not None: # base_url = server_url # else: # base_url = self._get_url(base_url, url_variables) # if not isinstance(request, BaseModel): # request = utils.unmarshal( # request, operations.GetEncryptionCertificateRequest # ) # request = cast(operations.GetEncryptionCertificateRequest, request) # req = self._build_request( # method="GET", # path="/api/v1/users/secrets/encryption-certificate", # base_url=base_url, # url_variables=url_variables, # request=request, # request_body_required=False, # request_has_path_params=False, # request_has_query_params=True, # user_agent_header="user-agent", # accept_header_value="application/json", # http_headers=http_headers, # security=self.sdk_configuration.security, # timeout_ms=timeout_ms, # ) # if retries == UNSET: # if self.sdk_configuration.retry_config is not UNSET: # retries = self.sdk_configuration.retry_config # else: # retries = utils.RetryConfig( # "backoff", utils.BackoffStrategy(3000, 720000, 1.88, 1800000), True # ) # retry_config = None # if isinstance(retries, utils.RetryConfig): # retry_config = (retries, ["5xx"]) # http_res = self.do_request( # hook_ctx=HookContext( # config=self.sdk_configuration, # base_url=base_url or "", # operation_id="get_encryption_certificate", # oauth2_scopes=[], # security_source=self.sdk_configuration.security, # ), # request=req, # error_status_codes=["422", "4XX", "5XX"], # retry_config=retry_config, # ) # response_data: Any = None # if utils.match_response(http_res, "200", "application/json"): # return operations.GetEncryptionCertificateResponse( # encryption_certificate_response=unmarshal_json_response( # Optional[shared.EncryptionCertificateResponse], http_res # ), # status_code=http_res.status_code, # content_type=http_res.headers.get("Content-Type") or "", # raw_response=http_res, # ) # if utils.match_response(http_res, "422", "application/json"): # response_data = unmarshal_json_response( # errors.HTTPValidationErrorData, http_res # ) # raise errors.HTTPValidationError(response_data, http_res) # if utils.match_response(http_res, "4XX", "*"): # http_res_text = utils.stream_to_text(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # if utils.match_response(http_res, "5XX", "*"): # http_res_text = utils.stream_to_text(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # raise errors.SDKError("Unexpected response received", http_res) # async def get_encryption_certificate_async( # self, # *, # request: Union[ # operations.GetEncryptionCertificateRequest, # operations.GetEncryptionCertificateRequestTypedDict, # ], # retries: OptionalNullable[utils.RetryConfig] = UNSET, # server_url: Optional[str] = None, # timeout_ms: Optional[int] = None, # http_headers: Optional[Mapping[str, str]] = None, # ) -> operations.GetEncryptionCertificateResponse: # r"""Retrieve the user's public key for encryption. # Retrieve a short lived certificate with the public key for encrypting secrets. # :param request: The request object to send. # :param retries: Override the default retry configuration for this method # :param server_url: Override the default server URL for this method # :param timeout_ms: Override the default request timeout configuration for this method in milliseconds # :param http_headers: Additional headers to set or replace on requests. # """ # base_url = None # url_variables = None # if timeout_ms is None: # timeout_ms = self.sdk_configuration.timeout_ms # if server_url is not None: # base_url = server_url # else: # base_url = self._get_url(base_url, url_variables) # if not isinstance(request, BaseModel): # request = utils.unmarshal( # request, operations.GetEncryptionCertificateRequest # ) # request = cast(operations.GetEncryptionCertificateRequest, request) # req = self._build_request_async( # method="GET", # path="/api/v1/users/secrets/encryption-certificate", # base_url=base_url, # url_variables=url_variables, # request=request, # request_body_required=False, # request_has_path_params=False, # request_has_query_params=True, # user_agent_header="user-agent", # accept_header_value="application/json", # http_headers=http_headers, # security=self.sdk_configuration.security, # timeout_ms=timeout_ms, # ) # if retries == UNSET: # if self.sdk_configuration.retry_config is not UNSET: # retries = self.sdk_configuration.retry_config # else: # retries = utils.RetryConfig( # "backoff", utils.BackoffStrategy(3000, 720000, 1.88, 1800000), True # ) # retry_config = None # if isinstance(retries, utils.RetryConfig): # retry_config = (retries, ["5xx"]) # http_res = await self.do_request_async( # hook_ctx=HookContext( # config=self.sdk_configuration, # base_url=base_url or "", # operation_id="get_encryption_certificate", # oauth2_scopes=[], # security_source=self.sdk_configuration.security, # ), # request=req, # error_status_codes=["422", "4XX", "5XX"], # retry_config=retry_config, # ) # response_data: Any = None # if utils.match_response(http_res, "200", "application/json"): # return operations.GetEncryptionCertificateResponse( # encryption_certificate_response=unmarshal_json_response( # Optional[shared.EncryptionCertificateResponse], http_res # ), # status_code=http_res.status_code, # content_type=http_res.headers.get("Content-Type") or "", # raw_response=http_res, # ) # if utils.match_response(http_res, "422", "application/json"): # response_data = unmarshal_json_response( # errors.HTTPValidationErrorData, http_res # ) # raise errors.HTTPValidationError(response_data, http_res) # if utils.match_response(http_res, "4XX", "*"): # http_res_text = await utils.stream_to_text_async(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # if utils.match_response(http_res, "5XX", "*"): # http_res_text = await utils.stream_to_text_async(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # raise errors.SDKError("Unexpected response received", http_res) # def store_secret( # self, # *, # request: Union[ # operations.StoreSecretRequest, operations.StoreSecretRequestTypedDict # ], # retries: OptionalNullable[utils.RetryConfig] = UNSET, # server_url: Optional[str] = None, # timeout_ms: Optional[int] = None, # http_headers: Optional[Mapping[str, str]] = None, # ) -> operations.StoreSecretResponse: # r"""Store an encrypted secret # After encrypting a secret locally, store it and get back a reference id. # :param request: The request object to send. # :param retries: Override the default retry configuration for this method # :param server_url: Override the default server URL for this method # :param timeout_ms: Override the default request timeout configuration for this method in milliseconds # :param http_headers: Additional headers to set or replace on requests. # """ # base_url = None # url_variables = None # if timeout_ms is None: # timeout_ms = self.sdk_configuration.timeout_ms # if server_url is not None: # base_url = server_url # else: # base_url = self._get_url(base_url, url_variables) # if not isinstance(request, BaseModel): # request = utils.unmarshal(request, operations.StoreSecretRequest) # request = cast(operations.StoreSecretRequest, request) # req = self._build_request( # method="POST", # path="/api/v1/users/secrets", # base_url=base_url, # url_variables=url_variables, # request=request, # request_body_required=True, # request_has_path_params=False, # request_has_query_params=True, # user_agent_header="user-agent", # accept_header_value="application/json", # http_headers=http_headers, # security=self.sdk_configuration.security, # get_serialized_body=lambda: utils.serialize_request_body( # request.encrypted_secret, False, False, "json", shared.EncryptedSecret # ), # timeout_ms=timeout_ms, # ) # if retries == UNSET: # if self.sdk_configuration.retry_config is not UNSET: # retries = self.sdk_configuration.retry_config # else: # retries = utils.RetryConfig( # "backoff", utils.BackoffStrategy(3000, 720000, 1.88, 1800000), True # ) # retry_config = None # if isinstance(retries, utils.RetryConfig): # retry_config = (retries, ["5xx"]) # http_res = self.do_request( # hook_ctx=HookContext( # config=self.sdk_configuration, # base_url=base_url or "", # operation_id="store_secret", # oauth2_scopes=[], # security_source=self.sdk_configuration.security, # ), # request=req, # error_status_codes=["422", "4XX", "5XX"], # retry_config=retry_config, # ) # response_data: Any = None # if utils.match_response(http_res, "200", "application/json"): # return operations.StoreSecretResponse( # secret_reference=unmarshal_json_response( # Optional[shared.SecretReference], http_res # ), # status_code=http_res.status_code, # content_type=http_res.headers.get("Content-Type") or "", # raw_response=http_res, # ) # if utils.match_response(http_res, "422", "application/json"): # response_data = unmarshal_json_response( # errors.HTTPValidationErrorData, http_res # ) # raise errors.HTTPValidationError(response_data, http_res) # if utils.match_response(http_res, "4XX", "*"): # http_res_text = utils.stream_to_text(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # if utils.match_response(http_res, "5XX", "*"): # http_res_text = utils.stream_to_text(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # raise errors.SDKError("Unexpected response received", http_res) # async def store_secret_async( # self, # *, # request: Union[ # operations.StoreSecretRequest, operations.StoreSecretRequestTypedDict # ], # retries: OptionalNullable[utils.RetryConfig] = UNSET, # server_url: Optional[str] = None, # timeout_ms: Optional[int] = None, # http_headers: Optional[Mapping[str, str]] = None, # ) -> operations.StoreSecretResponse: # r"""Store an encrypted secret # After encrypting a secret locally, store it and get back a reference id. # :param request: The request object to send. # :param retries: Override the default retry configuration for this method # :param server_url: Override the default server URL for this method # :param timeout_ms: Override the default request timeout configuration for this method in milliseconds # :param http_headers: Additional headers to set or replace on requests. # """ # base_url = None # url_variables = None # if timeout_ms is None: # timeout_ms = self.sdk_configuration.timeout_ms # if server_url is not None: # base_url = server_url # else: # base_url = self._get_url(base_url, url_variables) # if not isinstance(request, BaseModel): # request = utils.unmarshal(request, operations.StoreSecretRequest) # request = cast(operations.StoreSecretRequest, request) # req = self._build_request_async( # method="POST", # path="/api/v1/users/secrets", # base_url=base_url, # url_variables=url_variables, # request=request, # request_body_required=True, # request_has_path_params=False, # request_has_query_params=True, # user_agent_header="user-agent", # accept_header_value="application/json", # http_headers=http_headers, # security=self.sdk_configuration.security, # get_serialized_body=lambda: utils.serialize_request_body( # request.encrypted_secret, False, False, "json", shared.EncryptedSecret # ), # timeout_ms=timeout_ms, # ) # if retries == UNSET: # if self.sdk_configuration.retry_config is not UNSET: # retries = self.sdk_configuration.retry_config # else: # retries = utils.RetryConfig( # "backoff", utils.BackoffStrategy(3000, 720000, 1.88, 1800000), True # ) # retry_config = None # if isinstance(retries, utils.RetryConfig): # retry_config = (retries, ["5xx"]) # http_res = await self.do_request_async( # hook_ctx=HookContext( # config=self.sdk_configuration, # base_url=base_url or "", # operation_id="store_secret", # oauth2_scopes=[], # security_source=self.sdk_configuration.security, # ), # request=req, # error_status_codes=["422", "4XX", "5XX"], # retry_config=retry_config, # ) # response_data: Any = None # if utils.match_response(http_res, "200", "application/json"): # return operations.StoreSecretResponse( # secret_reference=unmarshal_json_response( # Optional[shared.SecretReference], http_res # ), # status_code=http_res.status_code, # content_type=http_res.headers.get("Content-Type") or "", # raw_response=http_res, # ) # if utils.match_response(http_res, "422", "application/json"): # response_data = unmarshal_json_response( # errors.HTTPValidationErrorData, http_res # ) # raise errors.HTTPValidationError(response_data, http_res) # if utils.match_response(http_res, "4XX", "*"): # http_res_text = await utils.stream_to_text_async(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # if utils.match_response(http_res, "5XX", "*"): # http_res_text = await utils.stream_to_text_async(http_res) # raise errors.SDKError("API error occurred", http_res, http_res_text) # raise errors.SDKError("Unexpected response received", http_res)