diff --git a/.gitignore b/.gitignore index 1329348..a1d83ab 100644 --- a/.gitignore +++ b/.gitignore @@ -116,3 +116,5 @@ dmypy.json # Pycharm .idea + +*.deb diff --git a/src/wakemebot/aptly.py b/src/wakemebot/aptly.py index 637df13..338b6b7 100644 --- a/src/wakemebot/aptly.py +++ b/src/wakemebot/aptly.py @@ -1,3 +1,4 @@ +import base64 import sys import uuid from contextlib import contextmanager @@ -6,7 +7,8 @@ from itertools import groupby from operator import attrgetter from pathlib import Path -from typing import Callable, Iterator, List, Set +from tempfile import TemporaryDirectory +from typing import Callable, Generator, Iterator, List, Set import httpx from debian.debian_support import Version, version_compare @@ -38,9 +40,23 @@ def sort_cmp(p1: Package, p2: Package) -> int: return 0 -def client_factory(server: str) -> httpx.Client: - transport = httpx.HTTPTransport(uds=server, retries=2) - return httpx.Client(transport=transport, base_url="http://aptly/api", timeout=30) +@contextmanager +def client_factory( + server_url: str, ca_cert: str, client_cert: str, client_key: str +) -> Generator[httpx.Client, None, None]: + with TemporaryDirectory(prefix="wakemebot_") as base_directory: + ca_cert_path = Path(base_directory) / "ca.crt" + ca_cert_path.write_bytes(base64.b64decode(ca_cert)) + client_cert_path = Path(base_directory) / "client.crt" + client_cert_path.write_bytes(base64.b64decode(client_cert)) + client_key_path = Path(base_directory) / "client.key" + client_key_path.write_bytes(base64.b64decode(client_key)) + transport = httpx.HTTPTransport( + retries=2, + cert=(str(client_cert_path), str(client_key_path)), + verify=str(ca_cert_path), + ) + yield httpx.Client(transport=transport, base_url=server_url, timeout=30) def parse_packages(data: List[str]) -> List[Package]: @@ -150,9 +166,15 @@ def upload_packages(client: httpx.Client, packages: List[Path], repo: str) -> No response.raise_for_status() -def push(repo: str, package_directory: Path, retain: int, server: str) -> None: - client = client_factory(server) - +def push( + repo: str, + package_directory: Path, + retain: int, + server_url: str, + ca_cert: str, + client_cert: str, + client_key: str, +) -> None: if package_directory.is_dir() is False: print(f"{package_directory} is not a directory") sys.exit(1) @@ -163,13 +185,33 @@ def push(repo: str, package_directory: Path, retain: int, server: str) -> None: if not packages: return - repos = [r["Name"] for r in client.get("/repos").json()] - - if repo not in repos: - print(f"Aptly repository {repo} not found.") - return - - upload_packages(client, packages, repo) - names = {file.name.split("_")[0] for file in packages} - for repo in repos: - purge(client, repo, names, retain) + with client_factory(server_url, ca_cert, client_cert, client_key) as client: + response = client.get("/repos") + response.raise_for_status() + repos = [r["Name"] for r in response.json()] + if repo not in repos: + print(f"Aptly repository {repo} not found.") + return + upload_packages(client, packages, repo) + names = {file.name.split("_")[0] for file in packages} + for repo in repos: + purge(client, repo, names, retain) + + +def publish( + repo: str, + server_url: str, + ca_cert: str, + client_cert: str, + client_key: str, +) -> None: + with client_factory(server_url, ca_cert, client_cert, client_key) as client: + response = client.put( + f"/publish/{repo}", + json={ + "ForceOverwrite": True, + "Signing": {"GpgKey": "wakemebot@protonmail.com", "Batch": True}, + }, + ) + response.raise_for_status() + print(response.json()) diff --git a/src/wakemebot/cli.py b/src/wakemebot/cli.py index 990d34a..db62064 100644 --- a/src/wakemebot/cli.py +++ b/src/wakemebot/cli.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import Optional import typer from pydantic import BaseModel, Field, HttpUrl @@ -21,6 +22,41 @@ def version() -> None: typer.secho(__version__) +def client_configuration_callback(value: Optional[str]) -> str: + if value is None: + raise typer.BadParameter("Option must be set through env or cli") + return value + + +option_server_url: str = typer.Option( + None, + help="Aptly server URL", + envvar="WAKEMEBOT_APTLY_SERVER_URL", + callback=client_configuration_callback, +) + +option_ca_cert: str = typer.Option( + None, + help="Base64 encoded CA certificate", + envvar="WAKEMEBOT_APTLY_CA_CERT", + callback=client_configuration_callback, +) + +option_client_cert: str = typer.Option( + None, + help="Base64 encoded client certificate", + envvar="WAKEMEBOT_APTLY_CLIENT_CERT", + callback=client_configuration_callback, +) + +option_client_key: str = typer.Option( + None, + help="Base64 encoded client key", + envvar="WAKEMEBOT_APTLY_CLIENT_KEY", + callback=client_configuration_callback, +) + + @aptly_app.command(name="push", help="Push debian sources packages to aptly repository") def aptly_push( repository: str = typer.Argument(..., help="Aptly repository name"), @@ -30,11 +66,37 @@ def aptly_push( retain: int = typer.Option( 100, help="For each package, how many versions will be kept" ), - server: str = typer.Option( - "/var/lib/aptly/aptly.sock", help="Path to server unix socket" - ), + server_url: str = option_server_url, + ca_cert: str = option_ca_cert, + client_cert: str = option_client_cert, + client_key: str = option_client_key, +) -> None: + aptly.push( + repository, + package_directory, + retain, + server_url, + ca_cert, + client_cert, + client_key, + ) + + +@aptly_app.command(name="publish", help="Publish aptly repository") +def aptly_publish( + repository: str = typer.Argument(..., help="Aptly repository name"), + server_url: str = option_server_url, + ca_cert: str = option_ca_cert, + client_cert: str = option_client_cert, + client_key: str = option_client_key, ) -> None: - aptly.push(repository, package_directory, retain, server) + aptly.publish( + repository, + server_url, + ca_cert, + client_cert, + client_key, + ) app.add_typer(aptly_app, name="aptly")