Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixes uses of type :str = None to be :Optional[str] = None #9

Merged
merged 1 commit into from
Nov 20, 2024
Merged

Fixes uses of type :str = None to be :Optional[str] = None #9

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 34 additions & 34 deletions src/phase/utils/phase_io.py
View file
Open in desktop
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
Loading