Skip to content

Commit

Permalink
add vault config to simplify parameter passing
Browse files Browse the repository at this point in the history
  • Loading branch information
@Esgrove
Esgrove committed Nov 15, 2024
1 parent b8f35a6 commit 183c562
Show file tree
Hide file tree
Showing 4 changed files with 223 additions and 516 deletions.
138 changes: 22 additions & 116 deletions python-pyo3/python/n_vault/vault.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,17 @@ def __init__(
self.vault_region = vault_region
self.profile = profile

self.config = nitor_vault_rs.VaultConfig(
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)

def all(self) -> str:
"""
Return a string with all keys separated by os.linesep.
Expand All @@ -92,51 +103,21 @@ def delete(self, name: str) -> None:
"""
Delete data in S3 for given key name.
"""
return nitor_vault_rs.delete(
name,
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.delete(name, self.config)

def delete_many(self, names: Collection[str]) -> None:
"""
Delete data for multiple keys.
Takes in a collection of key name strings, such as a `list`, `tuple`, or `set`.
"""
return nitor_vault_rs.delete_many(
sorted(names),
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.delete_many(sorted(names), self.config)

def direct_decrypt(self, encrypted_data: bytes) -> bytes:
"""
Decrypt data with KMS.
"""
return nitor_vault_rs.direct_decrypt(
encrypted_data,
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.direct_decrypt(encrypted_data, self.config)

def direct_encrypt(self, data: Union[bytes, str]) -> bytes:
"""
Expand All @@ -145,35 +126,15 @@ def direct_encrypt(self, data: Union[bytes, str]) -> bytes:
if isinstance(data, str):
data = data.encode("utf-8")

return nitor_vault_rs.direct_encrypt(
data,
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.direct_encrypt(data, self.config)

def exists(self, name: str) -> bool:
"""
Check if the given key name already exists in the S3 bucket.
Returns True if the key exists, False otherwise.
"""
return nitor_vault_rs.exists(
name,
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.exists(name, self.config)

def init(self) -> Union[StackCreated, CloudFormationStackData]:
"""
Expand All @@ -185,14 +146,7 @@ def init(self) -> Union[StackCreated, CloudFormationStackData]:
Returns a `StackCreated` if a new vault stack was initialized,
or `CloudFormationStackData` if it already exists.
"""
result = nitor_vault_rs.init(
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
result = nitor_vault_rs.init(self.config)
result_status = result.get("result")
if result_status == "CREATED":
return StackCreated(**result)
Expand All @@ -207,49 +161,21 @@ def list_all(self) -> list[str]:
Returns a list of key names.
"""
return nitor_vault_rs.list_all(
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.list_all(self.config)

def lookup(self, name: str) -> str:
"""
Lookup value for given key name.
Always returns a string, with binary data encoded in base64.
"""
return nitor_vault_rs.lookup(
name,
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.lookup(name, self.config)

def stack_status(self) -> CloudFormationStackData:
"""
Get vault Cloudformation stack status.
"""
data = nitor_vault_rs.stack_status(
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
data = nitor_vault_rs.stack_status(self.config)
return CloudFormationStackData(**data)

def store(self, name: str, data: Union[bytes, str]) -> None:
Expand All @@ -259,18 +185,7 @@ def store(self, name: str, data: Union[bytes, str]) -> None:
if isinstance(data, str):
data = data.encode("utf-8")

return nitor_vault_rs.store(
name,
data,
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
return nitor_vault_rs.store(name, data, self.config)

def update(self) -> Union[StackUpdated, CloudFormationStackData]:
"""
Expand All @@ -279,16 +194,7 @@ def update(self) -> Union[StackUpdated, CloudFormationStackData]:
Returns `StackUpdated` if the vault stack was updated to a new version,
or `CloudFormationStackData` if it is already up to date.
"""
result = nitor_vault_rs.update(
vault_stack=self.vault_stack,
region=self.vault_region,
bucket=self.vault_bucket,
key=self.vault_key,
prefix=self.vault_prefix,
profile=self.profile,
iam_id=self.vault_iam_id,
iam_secret=self.vault_iam_secret,
)
result = nitor_vault_rs.update(self.config)
result_status = result.get("result")
if result_status == "UPDATED":
return StackUpdated(**result)
Expand Down
Loading

0 comments on commit 183c562

Please sign in to comment.