AWS dynamic dispatcher with transient secrets

biome.jsonc

@@ -0,0 +1,38 @@
+{
+    "$schema": "https://biomejs.dev/schemas/2.4.6/schema.json",
+    "vcs": {
+        "enabled": false,
+        "clientKind": "git",
+        "useIgnoreFile": false
+    },
+    "files": {
+        "ignoreUnknown": false,
+        "includes": ["lib/html/**"]
+    },
+    "formatter": {
+        "enabled": true,
+        "indentStyle": "space",
+        "indentWidth": 4,
+        "lineWidth": 118
+    },
+    "linter": {
+        "enabled": true,
+        "rules": {
+            "recommended": true,
+            "style": {
+                "useBlockStatements": "error"
+            }
+        }
+    },
+    "javascript": {
+        "formatter": {
+            "quoteProperties": "preserve",
+            "quoteStyle": "double"
+        }
+    },
+    "html": {
+        "formatter": {
+            "enabled": true
+        }
+    }
+}

job-runner.toml

@@ -42,6 +42,7 @@ default-image = 'ghcr.io/cockpit-project/tasks:latest'
 # ]
 
 [logs]
+attach-journal = false
 driver='local'   # 's3' or 'local'
 
 [logs.s3]

lib/aio/amqp.py

@@ -0,0 +1,161 @@
+# Copyright (C) 2026 Red Hat, Inc.
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import asyncio
+import contextlib
+import logging
+import os
+import ssl
+from collections.abc import Mapping, Sequence
+from typing import Self
+
+import pika
+import pika.credentials
+from pika import BasicProperties
+from pika.adapters.asyncio_connection import AsyncioConnection
+from pika.channel import Channel
+from pika.spec import Basic
+
+logger = logging.getLogger(__name__)
+
+
+def _make_connection_params(credentials: Mapping[str, str]) -> pika.ConnectionParameters:
+    host, port = credentials['amqp-server'].split(':')
+
+    if host == 'localhost':
+        return pika.ConnectionParameters(
+            host,
+            int(port),
+            credentials=pika.credentials.PlainCredentials('guest', 'guest'),
+        )
+
+    with contextlib.ExitStack() as stack:
+
+        def memfd(name: str) -> str:
+            fd = os.memfd_create(name)
+            stack.callback(os.close, fd)
+            os.write(fd, credentials[name].encode())
+            return f'/proc/self/fd/{fd}'
+
+        context = ssl.create_default_context(cafile=memfd('ca.pem'))
+        context.load_cert_chain(
+            keyfile=memfd('amqp-client.key'),
+            certfile=memfd('amqp-client.pem'),
+        )
+        context.check_hostname = False
+
+    return pika.ConnectionParameters(
+        host,
+        int(port),
+        ssl_options=pika.SSLOptions(context, server_hostname=host),
+        credentials=pika.credentials.ExternalCredentials(),
+    )
+
+
+class Queue:
+    def __init__(
+        self, credentials: Mapping[str, str], queues: Sequence[str], consumer_priority: int | None = None,
+    ) -> None:
+        self._params = _make_connection_params(credentials)
+        self._queues = tuple(queues)
+        self._consumer_priority = consumer_priority
+        self._consumer_tags = tuple[str, ...]()
+        self._messages = asyncio.Queue[tuple[int, bytes] | Exception]()
+        self._connection: AsyncioConnection | None = None
+        self._channel: Channel | None = None
+
+    async def __aenter__(self) -> Self:
+        init_done: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+
+        def on_channel_closed(channel: Channel, reason: Exception) -> None:
+            logger.error('AMQP channel closed: %r %r', channel, reason)
+            self.close(reason)
+
+        def on_channel_opened(channel: Channel) -> None:
+            logger.debug('AMQP channel opened')
+            self._channel = channel
+            channel.add_on_close_callback(on_channel_closed)
+            channel.basic_qos(prefetch_count=1, global_qos=True)
+            for queue in self._queues:
+                logger.debug('declaring queue %r', queue)
+                channel.queue_declare(queue, durable=True, arguments={"x-max-priority": 9})
+            init_done.set_result(None)
+
+        def on_connection_open(connection: AsyncioConnection) -> None:
+            logger.debug('AMQP connection opened %r', connection)
+            self._connection = connection
+            connection.channel(on_open_callback=on_channel_opened)
+
+        def on_connection_open_error(connection: AsyncioConnection, error: Exception) -> None:
+            logger.error('AMQP connection failed: %r %r', connection, error)
+            init_done.set_exception(error)
+
+        def on_connection_closed(connection: AsyncioConnection, reason: Exception) -> None:
+            logger.error('AMQP closed: %r %r', connection, reason)
+            # We might get the close before or after we finished initializing
+            if not init_done.done():
+                init_done.set_exception(reason)
+            else:
+                self.close(reason)
+
+        AsyncioConnection(
+            self._params,
+            on_open_callback=on_connection_open,
+            on_open_error_callback=on_connection_open_error,
+            on_close_callback=on_connection_closed,
+        )
+
+        try:
+            await init_done
+        except Exception:
+            self._connection = None
+            self._channel = None
+            raise
+        return self
+
+    async def __aexit__(self, *_args: object) -> None:
+        self.close()
+
+    def close(self, reason: Exception | None = None) -> None:
+        self._channel = None
+        self._consumer_tags = ()
+        if self._connection is not None:
+            self._connection.close()
+            self._connection = None
+        if reason is not None:
+            self._messages.put_nowait(reason)
+
+    async def next_message(self) -> tuple[int, bytes]:
+        self.start_deliveries()
+        message = await self._messages.get()
+        if isinstance(message, Exception):
+            raise message
+        return message
+
+    def ack(self, delivery_tag: int) -> None:
+        if self._channel is not None:
+            self._channel.basic_ack(delivery_tag)
+            logger.debug('acked tag=%r', delivery_tag)
+
+    def start_deliveries(self) -> None:
+        if self._consumer_tags or self._channel is None:
+            return
+        arguments = {'x-priority': self._consumer_priority} if self._consumer_priority is not None else None
+        self._consumer_tags = tuple(
+            self._channel.basic_consume(queue, on_message_callback=self._on_message, arguments=arguments)
+            for queue in self._queues
+        )
+        logger.debug('consuming tags=%r', self._consumer_tags)
+
+    def stop_deliveries(self) -> None:
+        if self._channel is not None:
+            for tag in self._consumer_tags:
+                self._channel.basic_cancel(tag)
+                logger.debug('cancelled consumer tag=%r', tag)
+        self._consumer_tags = ()
+
+    def _on_message(
+        self, _channel: Channel, method: Basic.Deliver, _properties: BasicProperties | None, body: bytes
+    ) -> None:
+        logger.debug('received message tag=%r', method.delivery_tag)
+        self._messages.put_nowait((method.delivery_tag, body))

lib/aio/job.py

@@ -15,7 +15,6 @@
 
 import asyncio
 import contextlib
-import itertools
 import json
 import logging
 import os
@@ -88,6 +87,11 @@ async def run_container(job: Job, subject: Subject, ctx: JobContext, log: LogStr
 
         log.write(f'Using container image: {container_image}\n')
 
+        try:
+            secret_args = ctx.prepare_secrets(job.secrets, tmpdir / 'secrets')
+        except LookupError as exc:
+            raise Failure(str(exc)) from exc
+
         args = [
             *ctx.container_cmd, 'run',
             # we run arbitrary commands in that container, which aren't prepared for being pid 1; reap zombies
@@ -97,7 +101,7 @@ async def run_container(job: Job, subject: Subject, ctx: JobContext, log: LogStr
             *(f'--env={k}={v}' for k, v in job.env.items()),
             '--env=TEST_ATTACHMENTS=/var/tmp/attachments',
             f'--env=COCKPIT_CI_LOG_URL={log.url}',
-            *itertools.chain.from_iterable(args for name, args in ctx.secrets_args.items() if name in job.secrets),
+            *secret_args,
 
             container_image,
 
@@ -169,10 +173,21 @@ async def run_job(job: Job, ctx: JobContext) -> None:
         status = subject.forge.get_status(job.subject.repo, subject.sha, job.context, log.url)
         logger.info('Log: %s', log.url)
 
+        journal = ''
+        invocation_id = os.getenv("INVOCATION_ID")
+        if ctx.attach_journal and invocation_id:
+            async with spawn(
+                ["journalctl", f"_SYSTEMD_INVOCATION_ID={invocation_id}"],
+                stdout=asyncio.subprocess.PIPE,
+            ) as journalctl:
+                stdout, _stderr = await journalctl.communicate()
+                journal = stdout.decode()
+
         try:
             log.start(
                 f'{title}\n\n'
-                f'Running on: {platform.node()}\n\n'
+                f'Running on: {platform.node()} as {invocation_id or "(unknown invocation)"}\n'
+                f'{journal}\n'
                 f'Job({json.dumps(job, default=lambda obj: obj.__dict__, indent=4)})\n\n'
             )
             await status.post('pending', 'In progress')

lib/aio/jobcontext.py

@@ -19,7 +19,6 @@
 import os
 import re
 import sys
-import tempfile
 import tomllib
 from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
@@ -33,13 +32,15 @@
     JsonError,
     JsonObject,
     JsonValue,
+    get_bool,
     get_dict,
     get_nested,
     get_str,
     get_str_map,
     get_strv,
     json_merge_patch,
     load_external_files,
+    typechecked,
 )
 from .local import LocalLogDriver
 from .s3 import S3LogDriver
@@ -68,7 +69,7 @@
     target = parent / name
     value = obj[name]
     if isinstance(value, str):
         target.write_text(value)
     elif isinstance(value, Mapping):
         target.mkdir(parents=True, exist_ok=True)
         with get_nested(obj, name) as nested:
@@ -84,7 +85,6 @@
     logs: LogDriver
     _forges: dict[str, Forge]
     _default_forge: str
-    _secret_paths: dict[str, str]
 
     def load_config(self, path: Path, name: str, *, missing_ok: bool = False) -> None:
         logger.debug('Loading %s configuration from %s', name, path)
@@ -128,46 +128,51 @@
         else:
             self.load_config(Path(xdg_config_home('cockpit-dev/job-runner.toml')), 'user', missing_ok=True)
 
-    def expand_secret(self, arg: str) -> str:
-        def replace(m: re.Match[str]) -> str:
+    def prepare_secrets(self, names: Sequence[str], tmpdir: Path) -> Sequence[str]:
+        resolved: dict[str, str] = {}
+        all_args: list[str] = []
+        for name in names:
             try:
-                return self._secret_paths[m.group(1)]
+                secret_args = self._container_secrets[name]
             except KeyError:
-                raise JsonError(None, f"undefined secret '{m.group(0)}'") from None
-
-        return re.sub(r'%\{([^}]+)\}', replace, arg)
-
-    def expand_secrets(self, args: Sequence[str]) -> tuple[str, ...]:
-        return tuple(self.expand_secret(arg) for arg in args)
+                raise LookupError(f'no container.secrets.{name} entry') from None
+            for arg in secret_args:
+                for ref in re.findall(r'%\{([^}]+)\}', arg):
+                    if ref not in resolved:
+                        if ref in self._external_secrets:
+                            resolved[ref] = os.path.expanduser(self._external_secrets[ref])
+                        else:
+                            try:
+                                tmpdir.mkdir(parents=True, exist_ok=True)
+                                resolved[ref] = str(unpack_inline_secret(tmpdir, self._inline_secrets, ref))
+                            except KeyError:
+                                msg = f'container.secrets.{name} references %{{{ref}}} but no value is configured'
+                                raise LookupError(msg) from None
+                all_args.append(re.sub(r'%\{([^}]+)\}', lambda m: resolved[m.group(1)], arg))
+        return all_args
 
     async def __aenter__(self) -> Self:
         try:
-            # Build secrets mapping for %{name} substitution
             with get_nested(self.config, 'secrets') as secrets_section:
-                external = get_str_map(secrets_section, 'external')
-                secret_paths = {k: os.path.expanduser(v) for k, v in external.items()}
-
-                with get_nested(secrets_section, 'inline') as inline:
-                    if conflicts := secret_paths.keys() & inline.keys():
-                        msg = f"secret(s) defined in both 'external' and 'inline': {', '.join(sorted(conflicts))}"
-                        raise JsonError(inline, msg)
-                    if len(inline) > 0:
-                        tmpdir = Path(self.enter_context(tempfile.TemporaryDirectory()))
-                        secret_paths.update({
-                            name: str(unpack_inline_secret(tmpdir, inline, name)) for name in inline
-                        })
-
-            self._secret_paths = secret_paths
+                self._external_secrets = get_str_map(secrets_section, 'external')
+                self._inline_secrets = get_dict(secrets_section, 'inline')
+                if conflicts := self._external_secrets.keys() & self._inline_secrets.keys():
+                    msg = f"secret(s) defined in both 'external' and 'inline': {', '.join(sorted(conflicts))}"
+                    raise JsonError(secrets_section, msg)
 
             with get_nested(self.config, 'container') as container:
                 self.container_cmd = get_strv(container, 'command')
                 self.container_run_args = get_strv(container, 'run-args')
                 with get_nested(container, 'secrets') as secrets:
-                    # expand here to catch configuration errors early
-                    self.secrets_args = {name: self.expand_secrets(get_strv(secrets, name)) for name in secrets}
+                    self._container_secrets = {
+                        name: [
+                            typechecked(arg, str) for arg in typechecked(args, list)
+                        ] for name, args in secrets.items()
+                    }
                 self.default_image = get_str(container, 'default-image')
 
             with get_nested(self.config, 'logs') as logs:
+                self.attach_journal = get_bool(logs, 'attach-journal')
                 driver = get_str(logs, 'driver')
                 if driver not in LOG_DRIVERS:
                     sys.exit(f'Unknown log driver {driver}')