feat!: use https instead of unix socket to talk to aptly

.gitignore

@@ -116,3 +116,5 @@ dmypy.json
 
 # Pycharm
 .idea
+
+*.deb

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": "[email protected]", "Batch": True},
+            },
+        )
+        response.raise_for_status()
+        print(response.json())

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")