Skip to content

Commit

Permalink
Fixes uses of type :str = None to be :Optional[str] = None
Browse files Browse the repository at this point in the history
  • Loading branch information
drcongo committed Nov 20, 2024
1 parent 685f090 commit a7e0e27
Showing 1 changed file with 34 additions and 34 deletions.
68 changes: 34 additions & 34 deletions src/phase/utils/phase_io.py
Original file line number Diff line number Diff line change
@@ -1,23 +1,23 @@
import requests
from typing import Tuple
from typing import List, Dict
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple

import requests
from nacl.bindings import (
crypto_kx_server_session_keys,
)

from .const import pss_service_pattern, pss_user_pattern
from .crypto import CryptoUtils
from .misc import phase_get_context, tag_matches
from .network import (
fetch_phase_user,
create_phase_secrets,
delete_phase_secrets,
fetch_app_key,
fetch_wrapped_key_share,
fetch_phase_secrets,
create_phase_secrets,
fetch_phase_user,
fetch_wrapped_key_share,
update_phase_secrets,
delete_phase_secrets
)
from nacl.bindings import (
crypto_kx_server_session_keys,
)
from .crypto import CryptoUtils
from .const import __ph_version__, pss_user_pattern, pss_service_pattern
from .misc import phase_get_context, normalize_tag, tag_matches
from .secret_referencing import resolve_all_secrets


@dataclass
Expand Down Expand Up @@ -80,8 +80,8 @@ def auth(self):

return "Success"

except ValueError as err:
raise ValueError(f"Invalid Phase credentials")
except ValueError:
raise ValueError("Invalid Phase credentials")


def init(self):
Expand All @@ -90,12 +90,12 @@ def init(self):
# Ensure the response is OK
if response.status_code != 200:
raise ValueError(f"Request failed with status code {response.status_code}: {response.text}")

# Parse and return the JSON content
return response.json()


def create(self, key_value_pairs: List[Tuple[str, str]], env_name: str, app_name: str, path: str = '/', override_value: str = None) -> requests.Response:
def create(self, key_value_pairs: List[Tuple[str, str]], env_name: str, app_name: str, path: str = '/', override_value: Optional[str] = None) -> requests.Response:
"""
Create secrets in Phase KMS with support for specifying a path and overrides.
Expand Down Expand Up @@ -150,7 +150,7 @@ def create(self, key_value_pairs: List[Tuple[str, str]], env_name: str, app_name
return create_phase_secrets(self._token_type, self._app_secret.app_token, env_id, secrets, self._api_host)


def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag: str = None, path: str = '') -> List[Dict]:
def get(self, env_name: str, keys: List[str] = None, app_name: Optional[str] = None, tag: Optional[str] = None, path: str = '') -> List[Dict]:
"""
Get secrets from Phase KMS based on key and environment, with support for personal overrides,
optional tag matching, decrypting comments, and now including path support and key digest optimization.
Expand All @@ -165,7 +165,7 @@ def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag:
Returns:
List[Dict]: A list of dictionaries for all secrets in the environment that match the criteria, including their paths.
"""

user_response = fetch_phase_user(self._token_type, self._app_secret.app_token, self._api_host)
if user_response.status_code != 200:
raise ValueError(f"Request failed with status code {user_response.status_code}: {user_response.text}")
Expand Down Expand Up @@ -221,7 +221,7 @@ def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag:
"comment": decrypted_comment,
"path": secret.get("path", "/"),
"application": app_name,
"environment": env_name
"environment": env_name
}

# Only add the secret to results if the requested keys are not specified or the decrypted key is one of the requested keys.
Expand All @@ -231,10 +231,10 @@ def get(self, env_name: str, keys: List[str] = None, app_name: str = None, tag:
return results


def update(self, env_name: str, key: str, value: str = None, app_name: str = None, source_path: str = '', destination_path: str = None, override: bool = False, toggle_override: bool = False) -> str:
def update(self, env_name: str, key: str, value: Optional[str] = None, app_name: Optional[str] = None, source_path: str = '', destination_path: Optional[str] = None, override: bool = False, toggle_override: bool = False) -> str:
"""
Update a secret in Phase KMS based on key and environment, with support for source and destination paths.
Args:
env_name (str): The name (or partial name) of the desired environment.
key (str): The key for which to update the secret value.
Expand All @@ -244,11 +244,11 @@ def update(self, env_name: str, key: str, value: str = None, app_name: str = Non
destination_path (str, optional): The new path for the secret, if changing its location. If not provided, the path is not updated.
override (bool, optional): Whether to update an overridden secret value. Defaults to False.
toggle_override (bool, optional): Whether to toggle the override state between active and inactive. Defaults to False.
Returns:
str: A message indicating the outcome of the update operation.
"""

user_response = fetch_phase_user(self._token_type, self._app_secret.app_token, self._api_host)
if user_response.status_code != 200:
raise ValueError(f"Request failed with status code {user_response.status_code}: {user_response.text}")
Expand Down Expand Up @@ -301,10 +301,10 @@ def update(self, env_name: str, key: str, value: str = None, app_name: str = Non
# This prevents toggling an override on a secret that doesn't have one.
if "override" not in matching_secret or matching_secret["override"] is None:
raise OverrideNotFoundException(key)

# Retrieve the current override state. If the override is not active, it defaults to False.
current_override_state = matching_secret["override"].get("is_active", False)

# Prepare the payload to update the override status. The value of the override remains unchanged,
# but the isActive status is toggled.
secret_update_payload["override"] = {
Expand Down Expand Up @@ -338,20 +338,20 @@ def update(self, env_name: str, key: str, value: str = None, app_name: str = Non
return f"Error: Failed to update secret. HTTP Status Code: {response.status_code}"


def delete(self, env_name: str, keys_to_delete: List[str], app_name: str = None, path: str = None) -> List[str]:
def delete(self, env_name: str, keys_to_delete: List[str], app_name: Optional[str] = None, path: Optional[str] = None) -> List[str]:
"""
Delete secrets in Phase KMS based on keys and environment, with optional path support.
Args:
env_name (str): The name (or partial name) of the desired environment.
keys_to_delete (List[str]): The keys for which to delete the secrets.
app_name (str, optional): The name of the desired application.
path (str, optional): The path within which to delete the secrets. If specified, only deletes secrets within this path.
Returns:
List[str]: A list of keys that were not found and could not be deleted.
"""

user_response = fetch_phase_user(self._token_type, self._app_secret.app_token, self._api_host)
if user_response.status_code != 200:
raise ValueError(f"Request failed with status code {user_response.status_code}: {user_response.text}")
Expand All @@ -372,7 +372,7 @@ def delete(self, env_name: str, keys_to_delete: List[str], app_name: str = None,
keys_not_found = []
secrets_response = fetch_phase_secrets(self._token_type, self._app_secret.app_token, env_id, self._api_host, path=path)
secrets_data = secrets_response.json()

for key in keys_to_delete:
found = False
for secret in secrets_data:
Expand All @@ -388,9 +388,9 @@ def delete(self, env_name: str, keys_to_delete: List[str], app_name: str = None,

if secret_ids_to_delete:
delete_phase_secrets(self._token_type, self._app_secret.app_token, env_id, secret_ids_to_delete, self._api_host)

return keys_not_found


def decrypt(self, phase_ciphertext) -> str | None:
"""
Expand Down

0 comments on commit a7e0e27

Please sign in to comment.