Skip to content

feat: Add output_template configuration option and add a test module #13

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

Open
brunato wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: Add output_template configuration option and add a test module #13

Show file tree
Hide file tree
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
34 changes: 19 additions & 15 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -326,14 +326,16 @@ tcs-garr --environment stg init
-h, --help show this help message and exit
--id ID ID of the certificate to download.
--output-filename OUTPUT_FILENAME
Optional filename to save the certificate inside default output_folder.
--force, -f Force overwrite if the output file already exists.
Save the certificate inside default output_folder using a custom filename.
--save Save the certificate to a file inside default output_folder using a
filename automatically generated from Common Name and download type.
--force, -f Force overwrite if the output file already exists.
--download-type {pemBundle,certificate}
Type of download: 'pemBundle' or 'certificate'. Default is 'pemBundle'.
Type of download: 'pemBundle' or 'certificate'. Default is 'pemBundle'.
```

Replace `ID` with the ID of the certificate you wish to download. You can use
`pemBundle` or `certificate` as arguments for specific download formats.
Replace `ID` with the ID of the certificate you wish to download. For default the certificate is dumped
to standard output. You can use `pemBundle` or `certificate` as arguments for specific download formats.

The `download` command allows you to download certificates requested via API or ACME.

Expand Down Expand Up @@ -568,16 +570,17 @@ docker build -t tcs-garr:latest .

### Environment variables

| Name | Description | Default Value |
| -------------------- | ---------------------------------- | --------------------- |
| HARICA_USERNAME | Username for HARICA authentication | None |
| HARICA_PASSWORD | Password for HARICA authentication | None |
| HARICA_TOTP_SEED | TOTP seed for two-factor auth | None |
| HARICA_OUTPUT_FOLDER | Directory for output files | ~/harica_certificates |
| HARICA_HTTP_PROXY | HTTP Proxy | None |
| HARICA_HTTPS_PROXY | HTTPS Proxy | None |
| WEBHOOK_URL | Webhook URL | None |
| WEBHOOK_TYPE | Webhook Type | Slack |
| Name | Description | Default Value |
|------------------------|------------------------------------------------------|-------------------------|
| HARICA_USERNAME | Username for HARICA authentication | None |
| HARICA_PASSWORD | Password for HARICA authentication | None |
| HARICA_TOTP_SEED | TOTP seed for two-factor auth | None |
| HARICA_OUTPUT_FOLDER | Directory for output files | `~/harica_certificates` |
| HARICA_OUTPUT_TEMPLATE | String template to map the Common Name to a filename | None |
| HARICA_HTTP_PROXY | HTTP Proxy | None |
| HARICA_HTTPS_PROXY | HTTPS Proxy | None |
| WEBHOOK_URL | Webhook URL | None |
| WEBHOOK_TYPE | Webhook Type | Slack |

Info about
[webhook](https://github.com/ConsortiumGARR/tcs-garr?tab=readme-ov-file#webhook)
Expand All @@ -601,6 +604,7 @@ docker run --name tcs-garr \
-e HARICA_PASSWORD=${HARICA_PASSWORD} \
-e HARICA_TOTP_SEED=${HARICA_TOTP_SEED} \
-e HARICA_OUTPUT_FOLDER=${HARICA_OUTPUT_FOLDER} \
-e HARICA_FILENAME_TEMPLATE=${HARICA_FILENAME_TEMPLATE} \
-e HARICA_HTTP_PROXY=${HARICA_HTTP_PROXY} \
-e HARICA_HTTPS_PROXY=${HARICA_HTTPS_PROXY} \
-e HARICA_WEBHOOK_URL=${HARICA_WEBHOOK_URL} \
Expand Down
61 changes: 51 additions & 10 deletions tcs_garr/commands/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
from abc import ABC, abstractmethod
from functools import wraps
from pathlib import Path

from colorama import Fore, Style

from tcs_garr.harica_client import HaricaClient
from tcs_garr.logger import setup_logger
from tcs_garr.notifications import NotificationManager
from tcs_garr.utils import HaricaClientConfig
from tcs_garr.utils import HaricaClientConfig, OutputTemplate


def requires_any_role(*roles):
Expand Down Expand Up @@ -65,7 +66,7 @@ def wrapper(self, *args, **kwargs):
# Ensure we have a client with valid authentication
client = self.harica_client

# Check if the user has all of the required roles (AND logic)
# Check if the user has all the required roles (AND logic)
has_all_required_roles = all(client.has_role(role) for role in roles)

if not has_all_required_roles:
Expand Down Expand Up @@ -180,14 +181,8 @@ def check_required_role(self, client: HaricaClient):
exit(1)

@abstractmethod
def execute(self, args):
"""
Execute the command with the parsed arguments.

Args:
args: The parsed command arguments
"""
pass
def execute(self):
"""Execute the command with the parsed arguments."""

def call_webhook(self, cert_type, cn, cert_id=None):
webhook_url = self._harica_config.webhook_url
Expand All @@ -210,3 +205,49 @@ def call_webhook(self, cert_type, cn, cert_id=None):

except Exception as e:
self.logger.error(f"Error sending webhook via NotificationManager: {e}")

def get_output_folder(self):
"""
Retrieve the default output folder from the configuration.

Returns:
str: The output folder path from the configuration.
"""
# Load environment-specific configuration
return self.harica_config.output_folder

def get_output_filepath(self, filename, output_folder: str | None = None, **kwargs: str) -> Path:
"""
Return an absolute filepath for the provided filename, joining it with the output_folder.
If output_template option is configured and the provided argument starts with a FQDN, the
filename is substituted with a filename/filepath determined by configured template. In all
cases the resulting path must be a subfolder of output_folder.

Args:
filename: the output filename.
output_folder: the output folder path, if not provided use the configured output_folder.
kwargs: additional keyword arguments to provide in case of template substitution.
Returns:
An `pathlib.Path` object that represents an absolute filepath.
"""
if not isinstance(filename, str) or not (filename := filename.strip()):
raise TypeError("1st argument must be a not empty string")

base_path = Path(output_folder or self.get_output_folder())
if not base_path.is_absolute():
raise ValueError("Configuration error: output_folder must be an absolute path")

if len(Path(filename).parts) == 1 and (output_template := self.harica_config.output_template):
try:
template = OutputTemplate(output_template)
except KeyError as err:
msg = "Your configuration for 'output_template' is invalid: {}"
raise ValueError(msg.format(err))
else:
filepath = base_path.joinpath(template.get_filepath(filename, **kwargs))
else:
filepath = base_path.joinpath(filename)

if not filepath.is_relative_to(base_path):
raise ValueError(f"File path {filepath} is not relative to {base_path}")
return filepath
108 changes: 60 additions & 48 deletions tcs_garr/commands/download.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import base64
import importlib.resources as pkg_resources
import os
import pathlib
import warnings

from cryptography import x509
Expand Down Expand Up @@ -41,11 +42,18 @@ def configure_parser(self, parser):
# Add the argument to specify the certificate ID to download
parser.add_argument("--id", required=True, help="ID of the certificate to download.")

# Option for saving the certificate to the filesystem using a default filename
parser.add_argument(
"--save",
action="store_true",
default=False,
help="Save the certificate inside configured output_folder using a default filename.",
)
# Optional output filename for saving the certificate
parser.add_argument(
"--output-filename",
default=None,
help="Optional filename to save the certificate inside the default output folder.",
help="Save the certificate inside configured output_folder using a custom filename.",
)
# Add force flag to allow overwriting the file
parser.add_argument("--force", "-f", action="store_true", help="Force overwrite if the output file already exists.")
Expand All @@ -58,19 +66,6 @@ def configure_parser(self, parser):
help="Type of download: 'pemBundle' or 'certificate'. Default is 'pemBundle'.",
)

def get_output_folder(self):
"""
Retrieve the default output folder from the configuration.

Args:
args (argparse.Namespace): The command-line arguments passed to the command.

Returns:
str: The output folder path from the configuration.
"""
# Load environment-specific configuration
return self.harica_config.output_folder

def get_trusted_intermediates(self):
"""
Load all trusted intermediate certificates from the 'certs' folder.
Expand Down Expand Up @@ -217,48 +212,65 @@ def execute(self):
p7b_data = base64.b64decode(p7b_base64)

# Load and extract the certificates from the PKCS7 data
if p7b_data:
pkcs7_cert = pkcs7.load_der_pkcs7_certificates(p7b_data)
if pkcs7_cert:
certificates = pkcs7_cert
if p7b_data and (pkcs7_cert := pkcs7.load_der_pkcs7_certificates(p7b_data)):
certificates = pkcs7_cert

# Load trusted intermediates
trusted_intermediates = self.get_trusted_intermediates()
# Load trusted intermediates
trusted_intermediates = self.get_trusted_intermediates()

# Complete the certificate chain with trusted intermediates
complete_chain = self.complete_chain(certificates, trusted_intermediates)
# Complete the certificate chain with trusted intermediates
complete_chain = self.complete_chain(certificates, trusted_intermediates)

# Convert certificates to PEM format and join them into a single string
data_to_write = "".join(
cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") for cert in complete_chain
)
# Convert certificates to PEM format and join them into a single string
data_to_write = "".join(
cert.public_bytes(serialization.Encoding.PEM).decode("utf-8") for cert in complete_chain
)

# Optionally inspect the certificate chain
if not self.inspect_certificate_chain(complete_chain, trusted_intermediates):
self.logger.error("Certificate chain is not complete or valid.")
return
# Optionally inspect the certificate chain
if not self.inspect_certificate_chain(complete_chain, trusted_intermediates):
self.logger.error("Certificate chain is not complete or valid.")
return

if data_to_write:
# Determine the output folder from the config
output_folder = self.get_output_folder()
if not data_to_write:
# Handle case where no data is found for the given certificate ID
self.logger.error(f"No data found for certificate ID {self.args.id}.")

# If the output folder and filename are provided, save the certificate to a file
if output_folder and self.args.output_filename:
output_path = os.path.join(output_folder, self.args.output_filename)
elif not self.args.save and not self.args.output_filename:
# No options for save to file: print the certificate data
print(data_to_write)
else:
# Save the certificate file under the configured output_folder
output_folder = self.get_output_folder()

# Check if the file already exists, and handle the force flag for overwriting
if os.path.exists(output_path) and not self.args.force:
self.logger.error(f"File {output_path} already exists. Use --force to overwrite.")
# Save with a custom output filename
if self.args.output_filename:
output_filepath = pathlib.Path(output_folder).joinpath(self.args.output_filename)
else:
cn = certificate["dN"].strip().partition("CN=")[-1].strip()
for chunk in cn.split("."):
if not chunk.replace("-", "").isalnum():
self.logger.error(f"Certificate Common Name ({chunk}) is not valid.")
exit(1)

if self.args.download_type == "pemBundle":
filename = cn + "_fullchain.pem"
else:
# Write the certificate data to the file (binary or text based on data type)
with open(output_path, "wb" if isinstance(data_to_write, bytes) else "w") as cert_file:
cert_file.write(data_to_write)
self.logger.info(f"Certificate saved to {output_path}")
filename = cn + ".pem"

output_filepath = self.get_output_filepath(filename, base_path=output_folder)

# Check if the file already exists, and handle the force flag for overwriting
if output_filepath.is_dir():
self.logger.error(f"File {output_filepath} is a directory!")
elif output_filepath.exists() and not self.args.force:
self.logger.error(f"File {output_filepath} already exists. Use --force to overwrite.")
else:
# If no filename is provided, print the certificate data
print(data_to_write)
else:
# Handle case where no data is found for the given certificate ID
self.logger.error(f"No data found for certificate ID {self.args.id}.")
output_filepath.parent.mkdir(parents=True, exist_ok=True)

# Write the certificate data to the file (binary or text based on data type)
with output_filepath.open("wb" if isinstance(data_to_write, bytes) else "w") as cert_file:
cert_file.write(data_to_write)
self.logger.info(f"Certificate saved to {output_filepath}")

except CertificateNotApprovedException:
self.logger.error(f"Certificate with id {self.args.id} has not been approved yet. Retry later.")
25 changes: 24 additions & 1 deletion tcs_garr/commands/init.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

import tcs_garr.settings as settings
from tcs_garr.commands.base import BaseCommand
from tcs_garr.utils import OutputTemplate


class InitCommand(BaseCommand):
Expand Down Expand Up @@ -84,6 +85,7 @@ def _create_config_file(self, environment="production", force=False):
"password": "",
"totp_seed": "",
"output_folder": settings.OUTPUT_PATH,
"output_template": "",
"http_proxy": "",
"https_proxy": "",
"webhook_url": "",
Expand Down Expand Up @@ -158,6 +160,25 @@ def _create_config_file(self, environment="production", force=False):
# Expand in case input was a relative path
output_folder = os.path.abspath(os.path.expanduser(output_folder))

# Prompt for optional output filename setting
while True:
output_template_prompt = f"{Fore.GREEN}🗎 Enter output_template (optional)"
if force and has_existing_config and existing_values["output_template"]:
output_template_prompt += f" [{existing_values['output_template']}]"
output_template = input(f"{output_template_prompt}: {Style.RESET_ALL}") or (
existing_values["output_template"] if force and has_existing_config else ""
)

if output_template:
try:
OutputTemplate(output_template)
except ValueError:
retry_prompt = f"{Fore.YELLOW} Provided output_template is invalid! Retry? (Y/n)"
retry_option = input(f"{retry_prompt}: {Style.RESET_ALL}")
if retry_option.lower() == "y":
continue
break

# Prompt for optional proxy settings with existing values in brackets if force with existing config
http_proxy_prompt = f"{Fore.GREEN}🌐 Enter HTTP proxy (optional)"
if force and has_existing_config and existing_values["http_proxy"]:
Expand Down Expand Up @@ -206,7 +227,9 @@ def _create_config_file(self, environment="production", force=False):
"output_folder": output_folder,
}

# Add proxy settings only if they were provided
# Add optional settings only if they were provided
if output_template:
config[section_name]["output_template"] = output_template
if http_proxy:
config[section_name]["http_proxy"] = http_proxy
if https_proxy:
Expand Down
10 changes: 0 additions & 10 deletions tcs_garr/commands/k8s.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,16 +52,6 @@ def configure_parser(self, parser):
help="Name for the yaml file without the extension (optional).",
)

def get_output_folder(self):
"""
Retrieve the default output folder from the configuration.

Returns:
str: The output folder path from the configuration.
"""
# Load environment-specific configuration to get the output folder
return self.harica_config.output_folder

def execute(self):
"""
Executes the command to generate a Kubernetes TLS secret YAML file.
Expand Down
Loading