diff --git a/biome.jsonc b/biome.jsonc new file mode 100644 index 0000000000..df24ded148 --- /dev/null +++ b/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 + } + } +} diff --git a/job-runner.toml b/job-runner.toml index 3ab8e7f5fa..1e7c88b351 100644 --- a/job-runner.toml +++ b/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] diff --git a/lib/aio/amqp.py b/lib/aio/amqp.py new file mode 100644 index 0000000000..8fc72fc43f --- /dev/null +++ b/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)) diff --git a/lib/aio/job.py b/lib/aio/job.py index f419c231b6..c4541846e9 100644 --- a/lib/aio/job.py +++ b/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,22 @@ 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 = '' + # https://www.freedesktop.org/software/systemd/man/latest/systemd.exec.html#%24INVOCATION_ID + 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') diff --git a/lib/aio/jobcontext.py b/lib/aio/jobcontext.py index 1b48dee800..392f80384c 100644 --- a/lib/aio/jobcontext.py +++ b/lib/aio/jobcontext.py @@ -14,12 +14,12 @@ # along with this program. If not, see . import contextlib +import functools import json import logging import os import re import sys -import tempfile import tomllib from collections.abc import Callable, Mapping, Sequence from pathlib import Path @@ -33,6 +33,7 @@ JsonError, JsonObject, JsonValue, + get_bool, get_dict, get_nested, get_str, @@ -40,6 +41,7 @@ get_strv, json_merge_patch, load_external_files, + typechecked, ) from .local import LocalLogDriver from .s3 import S3LogDriver @@ -84,7 +86,6 @@ class JobContext(contextlib.AsyncExitStack): 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 +129,51 @@ def __init__(self, config_file: Path | str | None = None, *, debug: bool = False 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]: + def get_args(name: str) -> Sequence[str]: try: - return self._secret_paths[m.group(1)] + return self._container_secrets[name] except KeyError: - raise JsonError(None, f"undefined secret '{m.group(0)}'") from None + raise LookupError(f'no container.secrets.{name} entry') from None - return re.sub(r'%\{([^}]+)\}', replace, arg) + @functools.cache # cached until this function returns + def get_secret(secret_name: str) -> str: + if secret_name in self._external_secrets: + return os.path.expanduser(self._external_secrets[secret_name]) + try: + tmpdir.mkdir(parents=True, exist_ok=True) + return str(unpack_inline_secret(tmpdir, self._inline_secrets, secret_name)) + except KeyError: + raise LookupError(f'secret %{{{secret_name}}} is not configured') from None - def expand_secrets(self, args: Sequence[str]) -> tuple[str, ...]: - return tuple(self.expand_secret(arg) for arg in args) + return [ + re.sub(r'%\{(?P[^}]+)\}', lambda m: get_secret(m['secret']), arg) + for name in names + for arg in get_args(name) + ] 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}') diff --git a/lib/aws/README.md b/lib/aws/README.md new file mode 100644 index 0000000000..46ba6b8c80 --- /dev/null +++ b/lib/aws/README.md @@ -0,0 +1,321 @@ +# Cockpit CI AWS infrastructure guide + +## Files in this directory + +### Common + + - [`account.py`](account.py): contains constants like account numbers or S3 + bucket names that are used to configure and access various AWS services. + Things belong in this file if they're used to setup the infrastructure + (`infractl sync`) and also to *use* the infrastructure, for example the + names of S3 buckets where images are downloaded. In general, the constants + in this file are really useful with `git grep` to connect infrastructure + definitions to actual runtime uses. + +### Infrastructure deployment + + - [`authorized_keys`](authorized_keys): an `authorized_keys` file for who is + allowed to ssh to the dispatcher + - [`infra_definitions.py`](infra_definitions.py): the definition of our AWS + deployment. This uses many constants from [`account.py`](account.py) and + generally defines the shape of our infra in terms of IAM Policies, Roles, + Users, S3 Buckets, and EC2 launch templates and auto-scaling groups. It + also defines several SSM parameters which customize the dispatcher and + runners and contain secrets. + - [`ensure_resource.py`](ensure_resource.py): a set of helpers which allow + [`infra_definitions.py`](infra_definitions.py) to be written in a + declarative/idempotent style. Each function in here is more or less + responsible for making sure a particular resource exists, and has a + particular state. + - [`infractl.py`](infractl.py): a tool for performing various infrastructure + tasks. Invoke this as `python -m lib.aws.infractl` from the root of the + bots checkout. + +### Runtime/dispatcher + + - [`dashboard.html`](dashboard.html): an HTML published on the logs bucket + when the dispatcher is running. It fetches a file called `summary.json` + which contains the current set of active jobs and runners. + - [`dispatcher.py`](dispatcher.py): the core dispatcher logic. This is the part that consumes + jobs from the AMQP queue and launches EC2 instances. + - [`ec2.py`](ec2.py): helpers for launching and querying EC2 instances. This + is mostly used for runners, since the dispatcher itself is run from an + auto-scaling group. Launching needs a job-runner config and a job. + - [`jobconfig.py`](jobconfig.py): a generator for a JSON form of a + `job-runner.toml` containing ephemeral S3 credentials issued via the STS + service. This is used by the dispatcher to create ephemeral configurations + for runners but it could also be adapted for use with non-EC2 runners, or + even for local `job-runner`. + - [`launch_runner.py`](launch_runner.py): a utility script which can be used + for launching one-off runners for testing + +## Overview + +The CI deployment is "infrastructure as code". We went with Python instead of +other solutions for three main reasons: + + - Python is the most widely-understood language in the Cockpit team + + - the AWS Python bindings (`boto3`) are high quality, officially supported, + very actively maintained by Amazon, and with very good type annotations + available + + - we don't need any external "state" storage as is often required by other + solutions (OpenTofu, CloudFormation, etc). The deployed state of the + infrastructure itself is the only state. + +The infrastructure is completely defined inside of `lib/aws/account.py` (for +"constants' which are also shared by runtime code) and +`lib/aws/infra_definitions.py` (for the straight-up infra deployment). +Although those definitions are written as a series of Python functions, they +are written in a declarative/idempotent way, directly stating the desired state +of the deployment. + +There's a middle layer that takes the declarative form and converts it to +reality via AWS calls. That's the "helpers" in `lib/aws/ensure_resource.py`. +These are unloved, but relatively small and self-contained. It should +generally not be necessary to modify these unless adding new resource types. + + +## AWS on-boarding + +Everything that infractl does happens via boto3 which means that it understands +native AWS configuration (ie: configured via `~/.aws/`). In particular, you +may want to familiarize yourself with the [AWS +Documentation](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html), +particular around the use of the `AWS_PROFILE` environment variable. + +Make sure you have the `python3-boto3` package installed. The `awscli2` +package can also be very helpful (for the `aws` CLI tool). + +If you don't already have a `~/.aws/config`, here's a reasonable starting point: + +``` +[default] +region = us-east-1 +``` + +The credentials can be found in the `cockpit-infra-accounts` Bitwarden group. +Access to that group is available only on Red Hat Bitwarden accounts (search +the source for how to get an invite email) and is regulated via [a Rover +group](https://rover.redhat.com/groups/edit/basic/cockpit-infra-accounts). +Contact one of the owners of that group if you want to be added. + +You can add those credentials to `~/.aws/credentials` in a format similar to this: + +``` +[cockpit-ci-infractl] +aws_access_key_id = AKIA... +aws_secret_access_key = secret... +``` + +Some high-privilege operations below require access to the `admin` role on the +AWS account. These credentials are very powerful, and (accordingly) are not +associated with any static key, and are not stored in Bitwarden. They are only +available as an IAM role on time-limited sessions via SAML login and are +regulated by [a separate Rover +group](https://rover.redhat.com/groups/group/it-cloud-aws-727920394381-admin). +Contact the admins of that group if you need access. + +The easiest way to get this working is by configuring an `admin` profile in +`~/.aws/credentials`, something like so: + +``` +[admin] +credential_process = /home/lis/src/bots/main/saml-login --output=awscreds 727920394381-admin +``` + +with the `/home/lis/src/bots/main` path adjusted to point to your bots +checkout. + +You can test that your admin (or any other profile) access is working by using +a command like this: + +``` +# 'aws' from the awscli2 package +AWS_PROFILE=admin aws sts get-caller-identity +``` + +Finally, if you want to login to the AWS web console, you need to use this URL: +https://auth.redhat.com/auth/realms/EmployeeIDP/protocol/saml/clients/itaws + +In general, the web console is good for poking around or doing experiments, but +changes to infrastructure should be made to `infra_definitions.py`. + + +## ci-secrets repository + +The [ci-secrets +repository](https://gitlab.cee.redhat.com/front-door-ci-wranglers/ci-secrets/) +contains the static secrets used by the CI infrastructure. Normally you'll +want to check that out in `$XDG_RUNTIME_DIR/ci-secrets`. + +Access to this repository is regulated by [a rover +group](https://rover.redhat.com/groups/group/front-door-ci) and is only +available from the Red Hat internal network (or via VPN). + +You can configure your SSH key on the `gitlab.cee.redhat.com` instance via +https://gitlab.cee.redhat.com/-/user_settings/ssh_keys. You'll want to use +"Red Hat SAML Login" if prompted. + +Once everything is setup you should be able to clone the secrets like so: + +``` +git clone git@gitlab.cee.redhat.com:front-door-ci-wranglers/ci-secrets $XDG_RUNTIME_DIR/ci-secrets +``` + +## Task-oriented cheat sheet + +This is a "if you want to do X, type Y" list of instructions. + +All commands are run from a `cockpit-project/bots` checkout and assume the AWS +on-boarding (above) has been completed. + +### Deploy the entire CI infrastructure from scratch + +Deploying the infrastructure from scratch involves creation/modification of IAM +roles and policies which is a very highly-privileged operation. You'll need to +use the "admin" profile for this. + +You'll also want a checkout of the `ci-secrets` repository, mentioned above. + +``` +AWS_PROFILE=admin python3 -m lib.aws.infractl sync \ + --secrets-dir $XDG_RUNTIME_DIR/ci-secrets/aws-secrets/dispatcher/ + --bots-ref main +``` + +The entire process should take well under a minute to complete. When it's done +deploying, the `sync` command will scan existing resources on the cluster with +`cockpit` in the name that were unaffected by the current deployment. The goal +of this is to find any cruft lying around. You may want to remove things or +add them to the "known" `UNMANAGED_RESOURCES` list in `account.py` with an +explanation for why they're there. + + +### Updating secrets + +There's currently no way to update only the secrets. In order to do this, just +redeploy the entire infrastructure from scratch (which is harmless and fast). + + +### Make changes to the CI infrastructure + +You can make the required changes by modifying `lib/aws/infra_definitions.py` and running + +``` +AWS_PROFILE=admin python3 -m lib.aws.infractl sync --bots-ref main +``` + +if you omit the `--secrets-dir` argument then the secrets will be left unmodified. + + +### Changing the deployed version of the dispatcher or runners + +The version of the `bots` repo that gets checked out on the dispatcher and +runners is controlled by the `--bots-ref` argument to `infractl sync`. It's +important to note that this reference is resolved at deployment time and +remains hardcoded as a sha in the deployed configuration. If you make changes +to the dispatcher or `job-runner` and want them to be used, you need to +explicitly update them. This also gives a mechanism to roll back to known-good +versions. + +It's possible to do this without requiring a full infrastructure deploy. This +is also a lower-privilege operation and can be performed using the +`cockpit-ci-infractl` IAM user from Bitwarden). + +``` +AWS_PROFILE=cockpit-ci-infractl python3 -m lib.aws.infractl dispatcher update --bots-ref=main +``` + +The reference can be a branch name or a raw sha. + +You can also pass `--only-dispatcher` or `--only-runner` to only update the +version used on the dispatcher or the runners (which might be particular useful +during rollbacks). + +In any case, you'll need to restart the dispatcher after it's done in order to +pick up the new dispatcher version. You'll also need to restart the dispatcher +in order to pick up the new runner version because the parameter is read at +dispatcher startup time and sent to the runners from the dispatcher. + +### Restart the dispatcher + +The dispatcher is brought online by bringing up an instance to run it. The +launch template for the dispatcher will pick the latest version of its +configured OS and pull the version of the bots repository as configured above. + +Once the dispatcher is running, it is never updated in place. The only thing +that can be done is to power it off, at which point the auto-scaling group will +bring up a new instance. This can be done via: + +``` +AWS_PROFILE=cockpit-ci-infractl python3 -m lib.aws.infractl dispatcher restart +``` + +### Start or stop the dispatcher + +``` +AWS_PROFILE=cockpit-ci-infractl python3 -m lib.aws.infractl dispatcher up +AWS_PROFILE=cockpit-ci-infractl python3 -m lib.aws.infractl dispatcher down +``` + +This will configure the auto-scaling group to have a desired capacity of 0 or +1, effectively controlling if the dispatcher is running or not. + +### Check dispatcher status + +``` +AWS_PROFILE=cockpit-ci-infractl python3 -m lib.aws.infractl dispatcher status +``` + +This will show the status of the auto-scaling group and any dispatcher +instances that were found (running, terminated, etc.). + +### SSH to the dispatcher + +You can ssh to (`admin@`) the dispatcher public IP directly (as discovered by +the `status` command) but there's also a convenience wrapper: + +``` +AWS_PROFILE=cockpit-ci-infractl python3 -m lib.aws.infractl dispatcher ssh +``` + +In order to do that, you'll need to have your ssh key in the +`lib/aws/authorized_keys` file. Note: this file is baked into the launch +template of the dispatcher as part of `infractl sync` and `infractl dispatcher +update` will not update it. + +### List, inspect, or terminate runner instances + +``` +export AWS_PROFILE=cockpit-ci-infractl +python3 -m lib.aws.infractl runner list +python3 -m lib.aws.infractl runner list -a # include terminated +python3 -m lib.aws.infractl runner ssh SLUG +python3 -m lib.aws.infractl runner console SLUG # EC2 serial console (delayed ~10 min) +python3 -m lib.aws.infractl runner terminate SLUG +``` + +You can also check the [dashboard +page](https://cockpit-ci-logs.s3.us-east-1.amazonaws.com/dashboard.html) that +the dispatcher regularly updates when it's running: + +### Launch a one-off runner instance for debugging + +You can manually launch a runner instance for a given job JSON blob like so: + +``` +AWS_PROFILE=cockpit-ci python3 -m lib.aws.launch_runner '{"slug": "abc123", ...}' +``` + +If you omit the job then a synthetic job (which just sleeps) will be made up. + +In any case, you'll be connected to the machine via ssh as soon as it's +available. This is good for inspecting job runs and debugging issues. + +Since this involves launching EC2 instances (and requires access to the secrets +required to launch the instances) this requires the `cockpit-ci` profile from +Bitwarden. + +Closing the ssh connection will cause the instance to be terminated. diff --git a/lib/aws/__init__.py b/lib/aws/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/lib/aws/account.py b/lib/aws/account.py new file mode 100644 index 0000000000..adee4c8b0c --- /dev/null +++ b/lib/aws/account.py @@ -0,0 +1,102 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import timedelta +from typing import TYPE_CHECKING, Final, Literal + +if TYPE_CHECKING: + from types_boto3_s3.literals import BucketLocationConstraintType + +# This file defines constants for things like policy, role, bucket, etc. names +# along with some related constants like our AWS account number and SSO URLs. +# When adding new resources, ensure that 'cockpit-ci' appears in the name +# (ideally at the start) to make it easier to find resources that we've +# deployed. + +# It should be possible to find the resources with a command like: +# +# aws resource-explorer-2 search --query-string "cockpit" + +# ACCOUNT +# Things in this section are things that are to some extent outside of +# our control and may need to change. The account number is obvious but +# we may also need to change S3 bucket names since they are globally +# unique and first-come, first-serve. +ACCOUNT_ID = '727920394381' + +# S3 BUCKETS +# Bucket names are globally unique and first-come, first-serve. +LOGS_BUCKET = 'cockpit-ci-logs' +CI_IMAGES_BUCKETS: Mapping[str, BucketLocationConstraintType | Literal['us-east-1']] = { + 'cockpit-ci-images': 'us-east-1', + 'cockpit-ci-images-fra': 'eu-central-1', +} + +# TAGS +# Read corporate policy before changing anything here! +# https://source.redhat.com/departments/products_and_global_engineering/red_hat_public_cloud_services/public_cloud_services_wiki/resource_tagging_policy +# https://source.redhat.com/departments/products_and_global_engineering/red_hat_public_cloud_services/public_cloud_services_wiki/resource_tagging_names +MANDATORY_TAGS: Mapping[str, str] = { + 'app-code': 'ARR-001', + 'cost-center': '700', + 'service-phase': 'dev', +} +TAGS: Mapping[str, str] = { + **MANDATORY_TAGS, + 'service-owner': 'cockpit', # we've always done this... perhaps an older policy? +} + +# REGIONS +LOGS_REGION: Final = 'us-east-1' +# Also used for the dispatcher, SSM, STS, etc. +CI_RUNNER_REGION = 'us-east-1' + +# ROLES +# These are IAM roles that we've created. The names need only be unique +# inside of our AWS account, but we put them here as constants to avoid +# duplication. +DISPATCHER_ROLE = 'cockpit-ci-dispatcher' +IMAGE_DOWNLOAD_ROLE = 'cockpit-ci-images-download' +IMAGE_UPLOAD_ROLE = 'cockpit-ci-images-upload' +LOGS_WRITE_ROLE = 'cockpit-ci-logs-write' + +# The SAML download role is a separate role used by humans via Red Hat +# SSO (Rover). Rover requires the role name to start with our account +# ID. It shares the images-download managed policy with the CI role +# above but has a different trust policy (SAML vs same-account). +# Rover group: https://rover.redhat.com/groups/group/it-cloud-aws-727920394381-cockpit-ci-images-download +REDHAT_SSO_IDP_URL = 'https://auth.redhat.com/auth/realms/EmployeeIDP/protocol/saml/clients/itaws' +REDHAT_SSO_SAML_PROVIDER_ARN = f'arn:aws:iam::{ACCOUNT_ID}:saml-provider/RedHatInternal' +REDHAT_SSO_IMAGE_DOWNLOAD_ROLE = f'{ACCOUNT_ID}-cockpit-ci-images-download' +REDHAT_SSO_IMAGE_DOWNLOAD_MAX_SESSION = timedelta(hours=12) + +# DISPATCHER INSTANCE +DISPATCHER_ASG = 'cockpit-ci-dispatcher' +# Name used for the instance Name tag. +DISPATCHER_NAME = 'cockpit-ci/dispatcher' +# This is where we store config/secret parameters (in SSM) +DISPATCHER_PARAMS = '/cockpit-ci/dispatcher' + +# RUNNER INSTANCES +RUNNER_NAME_PREFIX = 'cockpit-ci/runner/' +RUNNER_INSTANCE_SLUG_TAG = 'cockpit-ci-slug' + +# SECURITY GROUPS +SSH_SECURITY_GROUP = 'cockpit-ci-ssh' + +# RESOURCE EXPLORER +# Query used to find all our resources, and ARNs that are expected but +# not managed by the bootstrap script. +RESOURCES_QUERY = 'cockpit' +UNMANAGED_RESOURCES: Mapping[str, str] = { + 'arn:aws:ec2:us-east-1:727920394381:elastic-ip/eipalloc-0234f925c7f590290': + 'cockpit-public-webhook elastic IP', + 'arn:aws:ec2:us-east-1:727920394381:network-interface/eni-004f5b4f714f3fda9': + 'cockpit-public-webhook ENI', +} + +# Useful derived constants +LOGS_URL = f'https://{LOGS_BUCKET}.s3.{LOGS_REGION}.amazonaws.com/' diff --git a/lib/aws/authorized_keys b/lib/aws/authorized_keys new file mode 100644 index 0000000000..7a1eee7b0e --- /dev/null +++ b/lib/aws/authorized_keys @@ -0,0 +1,6 @@ +sk-ssh-ed25519@openssh.com AAAAGnNrLXNzaC1lZDI1NTE5QG9wZW5zc2guY29tAAAAIB6oDRf2no5vPEM9ERg2n9ZT9Wpug/TAny/xjKgd+madAAAABHNzaDo= lis +ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIF8WnFWbcMzOSrQfBleOGLUSt52NfJt3oAMNbhRJGhxL lis-bitwarden-ansible + +ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBBpeQZYyFJYFgPhphjN68JUUKLgUUT63MxDui5DBiG/0Uj5qB9LRNyjN+7RP/LtkoqAcggTzD3fRjg95i+zxqvg= jelle@carbon + +ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIH1NqzyyTQvEPQqFy4TSmvFbtZNjVLFrYKAsAOTV3Mb+ tmatus@tmatus-rh diff --git a/lib/aws/dispatcher.py b/lib/aws/dispatcher.py new file mode 100644 index 0000000000..5d79e38b78 --- /dev/null +++ b/lib/aws/dispatcher.py @@ -0,0 +1,419 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +from collections import defaultdict +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import TYPE_CHECKING + +import boto3 +import botocore.exceptions +import httpx + +from ..aio.amqp import Queue +from ..aio.jsonutil import JsonObject, get_int, get_str, get_strv +from .account import CI_RUNNER_REGION, DISPATCHER_PARAMS, LOGS_BUCKET, LOGS_URL +from .ec2 import ( + describe_runner_instances, + get_instance_ip, + get_instance_slug, + get_instance_state, + launch_instance, +) +from .jobconfig import job_runner_config + +if TYPE_CHECKING: + from types_boto3_ec2 import EC2Client + from types_boto3_ec2.literals import InstanceStateNameType, InstanceTypeType + from types_boto3_sts import STSClient + +logger = logging.getLogger(__name__) + + +def load_parameters(source: str) -> dict[str, str]: + if source.startswith("ssm:"): + prefix = source.removeprefix("ssm:") + ssm = boto3.client("ssm", region_name=CI_RUNNER_REGION) + paginator = ssm.get_paginator("get_parameters_by_path") + pages = paginator.paginate(Path=prefix, WithDecryption=True, Recursive=True) + return { + param["Name"].removeprefix(prefix): param["Value"] + for page in pages + for param in page["Parameters"] + } + + if source.startswith("json:"): + return json.loads(source.removeprefix("json:")) + + if source.startswith("dir:"): + return { + p.name: p.read_text() + for p in sorted(Path(source.removeprefix("dir:")).iterdir()) + if p.is_file() + } + + raise ValueError(f"unknown parameter source: {source!r}") + + +def prepare_and_launch( + ec2: EC2Client, + sts: STSClient, + *, + job: JsonObject, + params: Mapping[str, str], + bots_url: str, + instance_type: InstanceTypeType, + post: bool, + ssh_keys: Sequence[str] = (), + ami: str | None = None, +) -> str: + slug = get_str(job, "slug") + job_timeout_min = min(get_int(job, "timeout", 120), MAX_JOB_TIMEOUT_MIN) + + # Three layers of timeouts, each giving the previous layer headroom: + # - job_timeout_min: how long the job runs before job-runner kills it + # - systemd_timeout_min: job timeout + 15 min for setup (download, uv, etc.) + # - credential_duration: job timeout + 30 min so credentials outlive the unit + # The instance hard kill (MAX_AGE_MIN) must be >= credential_duration. + systemd_timeout_min = job_timeout_min + 15 + credential_duration = timedelta(minutes=job_timeout_min + 30) + logger.debug( + "preparing job %r: timeout=%r systemd=%r credentials=%r", + slug, + job_timeout_min, + systemd_timeout_min, + credential_duration, + ) + + return launch_instance( + ec2, + bots_url=bots_url, + job={**job, "timeout": job_timeout_min}, + job_config=job_runner_config( + slug, + sts, + secrets=get_strv(job, "secrets", ()), + params=params, + post=post, + credential_duration=credential_duration, + ), + instance_type=instance_type, + systemd_timeout_min=systemd_timeout_min, + ami=ami, + ssh_keys=ssh_keys, + ) + + +@dataclass +class Instance: + instance_id: str + slug: str + state: "InstanceStateNameType" + launch_time: datetime + ip: str | None + + def to_json(self) -> JsonObject: + return { + "slug": self.slug, + "state": self.state, + "launch_time": self.launch_time.isoformat(), + "ip": self.ip, + } + + +class Job: + def __init__(self) -> None: + self.launched_instance: str | None = None + self.observed_instances = set[str]() + self.logs_visible: bool | None = None + self.human: str | None = None + + def should_check_logs(self, instances: Mapping[str, Instance]) -> bool: + if self.logs_visible is None: + return True + if self.logs_visible: + return False + return any( + instances.get(iid) is not None and instances[iid].state == "running" + for iid in self.observed_instances + ) + + def to_json(self) -> JsonObject: + return { + "launched_instance": self.launched_instance, + "observed_instances": sorted(self.observed_instances), + "logs_visible": self.logs_visible, + "human": self.human, + } + + +MAX_JOB_TIMEOUT_MIN = 120 +MAX_AGE_MIN = MAX_JOB_TIMEOUT_MIN + 30 + + +class Dispatcher: + def __init__( + self, + params: Mapping[str, str], + ssh_keys: Sequence[str] = (), + ) -> None: + self.params = params + self.ssh_keys = ssh_keys + self.instances: dict[str, Instance] = {} + self.jobs: defaultdict[str, Job] = defaultdict(Job) + self.logs_pending = asyncio.Event() + self.can_take_job = asyncio.Event() + + def check_capacity(self) -> bool: + # "Active" counts as: + # - any observed instance either pending or running + # - a job which has entered the system but hasn't been started + # - a job which has been started but not yet observed + n_active = sum(( + sum( + 1 + for inst in self.instances.values() + if inst.state in ("pending", "running") + ), + sum(1 for job in self.jobs.values() if not job.observed_instances), + )) + n_awaiting_logs = sum( + job.should_check_logs(self.instances) for job in self.jobs.values() + ) + + max_active = int(self.params.get("max-active", "50")) + max_awaiting_logs = int(self.params.get("max-awaiting-logs", "20")) + capacity = min(max_active - n_active, max_awaiting_logs - n_awaiting_logs) + if capacity > 0: + self.can_take_job.set() + return capacity > 0 + + async def ensure_job_running( + self, message: JsonObject, ec2: EC2Client, sts: STSClient + ) -> None: + job_json = message["job"] + assert isinstance(job_json, dict) + slug = job_json["slug"] + assert isinstance(slug, str) + job = self.jobs[slug] + job.human = get_str(message, "human", None) + loop = asyncio.get_running_loop() + + backoff = 15.0 + while not job.launched_instance and not job.observed_instances: + try: + job.launched_instance = await loop.run_in_executor( + None, + lambda: prepare_and_launch( + ec2, + sts, + job=job_json, + params=self.params, + bots_url=self.params["runner-url"], + instance_type="m8id.4xlarge", + post=True, + ssh_keys=self.ssh_keys, + ), + ) + self.logs_pending.set() + + except botocore.exceptions.ClientError as e: + code = e.response["Error"]["Code"] + if code not in ( + "InsufficientInstanceCapacity", + "RequestLimitExceeded", + "ServiceUnavailable", + ): + raise + logger.warning( + "launch failed for %r: %s, backing off %rs", slug, code, backoff + ) + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 300) + + async def ec2_launcher(self) -> None: + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + sts = boto3.client("sts", region_name=CI_RUNNER_REGION) + + async with Queue(self.params, queues=["public"], consumer_priority=10) as queue: + while await self.can_take_job.wait(): + delivery_tag, body = await queue.next_message() + + message = json.loads(body) + logger.info("got job %r", message.get("job", {}).get("slug")) + await self.ensure_job_running(message, ec2, sts) + + if not self.check_capacity(): + self.can_take_job.clear() + queue.stop_deliveries() + + queue.ack(delivery_tag) + + async def s3_observer(self) -> None: + async with httpx.AsyncClient() as http: + while await self.logs_pending.wait(): + unchecked = [ + (s, j) + for s, j in self.jobs.items() + if j.should_check_logs(self.instances) + ] + if not unchecked: + self.logs_pending.clear() + continue + + for slug, job in unchecked: + try: + resp = await http.head(f"{LOGS_URL}{slug}/log.html", timeout=5.) + job.logs_visible = resp.is_success + if resp.is_success: + logger.info("logs visible for %r", slug) + except httpx.HTTPError: + job.logs_visible = False + + self.check_capacity() + + await asyncio.sleep(10) + + async def scan_instances(self, ec2: "EC2Client") -> None: + loop = asyncio.get_running_loop() + + all_instances = await loop.run_in_executor(None, describe_runner_instances, ec2) + + self.instances = { + obj["InstanceId"]: Instance( + instance_id=obj["InstanceId"], + slug=get_instance_slug(obj), + state=get_instance_state(obj), + launch_time=obj["LaunchTime"].astimezone(timezone.utc), + ip=get_instance_ip(obj), + ) + for obj in all_instances + } + + now = datetime.now(timezone.utc) + + for inst in self.instances.values(): + job = self.jobs[inst.slug] + job.observed_instances.add(inst.instance_id) + if job.should_check_logs(self.instances): + self.logs_pending.set() + + overdue = [ + inst.instance_id + for inst in self.instances.values() + if inst.state in ("pending", "running") + and (now - inst.launch_time).total_seconds() > MAX_AGE_MIN * 60 + ] + if overdue: + logger.warning("terminating overdue instances: %r", overdue) + await loop.run_in_executor( + None, + lambda: ec2.terminate_instances(InstanceIds=overdue), + ) + + for slug in [ + s + for s, j in self.jobs.items() + if j.observed_instances and not j.observed_instances & self.instances.keys() + ]: + logger.info("pruning job %r", slug) + del self.jobs[slug] + + async def ec2_observer(self) -> None: + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + + while True: + await self.scan_instances(ec2) + + self.check_capacity() + + await asyncio.sleep(5) + + def to_json(self) -> JsonObject: + return { + "instances": {iid: inst.to_json() for iid, inst in self.instances.items()}, + "jobs": {slug: job.to_json() for slug, job in self.jobs.items()}, + } + + +async def main() -> None: + parser = argparse.ArgumentParser(description="EC2 CI job dispatcher") + # fmt: off + parser.add_argument("--debug", action="store_true") + parser.add_argument("--poll-interval", type=float, default=3) + parser.add_argument("--parameters", default=f"ssm:{DISPATCHER_PARAMS}/", + help="Parameter source: ssm:PREFIX, json:DATA, or dir:PATH") + parser.add_argument("--ssh-key", type=Path, + help="SSH public key file to authorize for the core user") + parser.add_argument("--param", action="append", default=[], + help="Override a parameter: --param key=value") + # fmt: on + args = parser.parse_args() + + logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO) + + loop = asyncio.get_running_loop() + s3 = boto3.client("s3", region_name=CI_RUNNER_REGION) + + def _upload_logs(key: str, body: str, mimetype: str) -> None: + logger.info("uploading %s/%s (%s)", LOGS_BUCKET, key, mimetype) + s3.put_object(Bucket=LOGS_BUCKET, Key=key, Body=body, ContentType=mimetype) + + dashboard_dir = Path(__file__).parent / "../html/dashboard" + for name, mimetype in [ + ("dashboard.html", "text/html"), + ("dashboard.js", "text/javascript") + ]: + await loop.run_in_executor( + None, _upload_logs, name, (dashboard_dir / name).read_text(), mimetype + ) + + params = load_parameters(args.parameters) + for override in args.param: + key, _, value = override.partition("=") + logger.debug("overriding parameter %r=%r", key, value) + params[key] = value + + ssh_keys = args.ssh_key.read_text().strip().splitlines() if args.ssh_key else () + dispatcher = Dispatcher(params=params, ssh_keys=ssh_keys) + + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + await dispatcher.scan_instances(ec2) + logger.info("initial scan: %d instances", len(dispatcher.instances)) + + # The dispatcher does four things pretty much all the time: + # - querying status of existing runners (the `ec2_observer` task) + # - listening for jobs to arrive from the queue and spawning EC2 instances + # (the `ec2_launcher` task) + # - querying S3 to see what logs are available (`s3_observer` task) + # - periodically updating the summary.json, from the main task + # + # The EC2 and S3 observers both influence how many "free slots" the + # dispatcher has for accepting new jobs. See `.check_capacity()`. + async with asyncio.TaskGroup() as tg: + tg.create_task(dispatcher.ec2_launcher()) + tg.create_task(dispatcher.ec2_observer()) + tg.create_task(dispatcher.s3_observer()) + + prev_summary = "" + while True: + summary = json.dumps(dispatcher.to_json(), default=str, indent=4) + if summary != prev_summary: + await loop.run_in_executor( + None, _upload_logs, "summary.json", summary, "application/json" + ) + prev_summary = summary + + await asyncio.sleep(args.poll_interval) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/lib/aws/ec2.py b/lib/aws/ec2.py new file mode 100644 index 0000000000..b6d69558fc --- /dev/null +++ b/lib/aws/ec2.py @@ -0,0 +1,317 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import base64 +import json +import logging +import sys +import time +import urllib.request +from collections.abc import Sequence +from typing import TYPE_CHECKING + +from ..aio.jsonutil import ( + JsonError, + JsonObject, + JsonValue, + get_nested, + get_str, + typechecked, +) + +if TYPE_CHECKING: + from types_boto3_ec2 import EC2Client + from types_boto3_ec2.literals import InstanceStateNameType, InstanceTypeType + from types_boto3_ec2.type_defs import InstanceTypeDef, TagTypeDef + +from .account import ( + RUNNER_INSTANCE_SLUG_TAG, + RUNNER_NAME_PREFIX, + SSH_SECURITY_GROUP, + TAGS, +) + +logger = logging.getLogger(__name__) + + +# listing +def describe_runner_instances( + ec2: EC2Client, *, slug: str = "*" +) -> Sequence[InstanceTypeDef]: + logger.debug("describing instances with slug=%r", slug) + paginator = ec2.get_paginator("describe_instances") + pages = paginator.paginate( + Filters=[{"Name": f"tag:{RUNNER_INSTANCE_SLUG_TAG}", "Values": [slug]}] + ) + result = [ + obj + for page in pages + for reservation in page["Reservations"] + for obj in reservation.get("Instances", ()) + ] + logger.debug("found %d instances", len(result)) + return result + + +def get_instance_slug(instance: InstanceTypeDef) -> str: + return next( + t["Value"] + for t in instance.get("Tags", ()) + if t["Key"] == RUNNER_INSTANCE_SLUG_TAG + ) + + +def get_instance_ip(instance: InstanceTypeDef) -> str | None: + return instance.get("PublicIpAddress") + + +def get_instance_state(instance: InstanceTypeDef) -> InstanceStateNameType: + return instance["State"]["Name"] + + +# launching +def _ensure_nested_virt_support(ec2: EC2Client) -> None: + # Debian carries an old botocore whose service model doesn't include the + # CpuOptions.NestedVirtualization parameter. The parameter is valid on the + # AWS side, so we patch the client's model to include it. + # This can be removed once the dispatcher runs on a distro with a current botocore. + shapes = ec2._service_model._shape_resolver._shape_map # type: ignore[attr-defined] + if "NestedVirtualization" not in shapes["CpuOptionsRequest"]["members"]: + logger.warning("patching CpuOptionsRequest to include NestedVirtualization") + shapes["CpuOptionsRequest"]["members"]["NestedVirtualization"] = { + "shape": "CpuOptionsNestedVirtualization" + } + shapes["CpuOptionsNestedVirtualization"] = {"type": "string"} + + +def string_contents(content: str) -> JsonObject: + return {"source": f"data:;base64,{base64.b64encode(content.encode()).decode()}"} + + +def json_contents(obj: JsonValue) -> JsonObject: + return string_contents(json.dumps(obj)) + + +_fcos_ami_cache: tuple[str, str] | None = None + + +def find_fcos_ami(region: str) -> str: + FCOS_STREAM_URL = "https://builds.coreos.fedoraproject.org/streams/stable.json" + global _fcos_ami_cache + + today = time.strftime("%Y-%m-%d") + if _fcos_ami_cache is None or _fcos_ami_cache[0] != today: + try: + logger.debug("fetching FCOS stream metadata for region %r", region) + with urllib.request.urlopen(FCOS_STREAM_URL) as response: + stream = typechecked(json.loads(response.read()), dict) + + with get_nested(stream, "architectures") as architectures: + with get_nested(architectures, "x86_64") as x86_64: + with get_nested(x86_64, "images") as images: + with get_nested(images, "aws") as aws: + with get_nested(aws, "regions") as regions: + with get_nested(regions, region) as entry: + ami = get_str(entry, "image") + + logger.debug("found FCOS AMI %r for region %r", ami, region) + _fcos_ami_cache = (today, ami) + + except (OSError, json.JSONDecodeError, JsonError): + if _fcos_ami_cache is None: + raise + logger.warning("failed to fetch FCOS stream metadata, using cached AMI %r", + _fcos_ami_cache[1], exc_info=True) + + return _fcos_ami_cache[1] + + +def resolve_security_group(ec2: EC2Client, name_or_id: str) -> str: + if name_or_id.startswith("sg-"): + return name_or_id + response = ec2.describe_security_groups( + Filters=[{"Name": "group-name", "Values": [name_or_id]}], + ) + groups = response["SecurityGroups"] + if not groups: + sys.exit(f"security group not found: {name_or_id!r}") + logger.debug("resolved security group %r to %r", name_or_id, groups[0]["GroupId"]) + return groups[0]["GroupId"] + + +def launch_instance( + ec2: EC2Client, + *, + bots_url: str, + job: JsonObject, + job_config: JsonObject, + instance_type: InstanceTypeType, + systemd_timeout_min: int, + ami: str | None = None, + ssh_keys: Sequence[str] = (), +) -> str: + _ensure_nested_virt_support(ec2) + + slug = job["slug"] + assert isinstance(slug, str) + + ignition = { + "ignition": {"version": "3.4.0"}, + "storage": { + "files": [ + { + "path": "/etc/systemd/zram-generator.conf", + "contents": string_contents( + "[zram0]\n" + "zram-size = ram / 2\n" + "compression-algorithm = zstd\n" + ), + "mode": 0o644, + }, + { + "path": "/etc/cockpit-ci/bots-url", + "contents": string_contents(bots_url), + "mode": 0o644, + }, + { + "path": "/etc/cockpit-ci/job-runner.json", + "contents": json_contents(job_config), + "mode": 0o644, + }, + { + "path": "/etc/cockpit-ci/job.json", + "contents": json_contents(job), + "mode": 0o644, + }, + { + "path": "/usr/local/bin/run-job", + "contents": string_contents(r"""#!/bin/bash + set -euxo pipefail + + maybe_sit() { + # If there are ssh keys configured, let the user inspect it + if [ -s /home/core/.ssh/authorized_keys.d/ignition ]; then + sleep 10m + fi + } + trap maybe_sit ERR + + # Set SELinux permissive + setenforce 0 + + # Block IMDS — ignition is done, nobody needs it, + # and the container shouldn't have access + iptables -A OUTPUT --destination 169.254.169.254 -j REJECT + + # Setup containers storage on local (fast) NVMe disk + instance_store_device=$( + lsblk -dpno NAME --filter 'MODEL=~"Instance Storage"' + ) + mkfs.btrfs -f "$instance_store_device" + mount "$instance_store_device" /var/lib/containers + + # Download bots and unpack it (we don't need git history) + bots_url="$( str: + return f"arn:aws:iam::{ACCOUNT_ID}:role/{name}" + + +def policy_document(statements: Sequence[JsonObject]) -> str: + return json.dumps( + {"Version": "2012-10-17", "Statement": statements}, sort_keys=True + ) + + +@contextlib.contextmanager +def _suppress( + code: str, + on_success: str = "created", + on_suppress: str = "already exists", +) -> Iterator[None]: + try: + yield + except ClientError as exc: + if exc.response["Error"]["Code"] != code: + raise + print(f" - {on_suppress}") + else: + print(f" - {on_success}") + + +def ensure_policy(name: str, statements: Sequence[JsonObject]) -> str: + iam = boto3.client("iam") + arn = f"arn:aws:iam::{ACCOUNT_ID}:policy/{name}" + + print(f" - policy {name}") + with _suppress("EntityAlreadyExists"): + iam.create_policy( + PolicyName=name, + PolicyDocument=policy_document(statements), + Tags=[{"Key": k, "Value": v} for k, v in TAGS.items()], + ) + + # AWS limits policies to 5 versions; delete non-defaults to make room + # before adding ours, then clean up the old default afterwards. + versions = iam.list_policy_versions(PolicyArn=arn)["Versions"] + old_default = next(v["VersionId"] for v in versions if v["IsDefaultVersion"]) + for v in versions: + if not v["IsDefaultVersion"]: + iam.delete_policy_version(PolicyArn=arn, VersionId=v["VersionId"]) + iam.create_policy_version( + PolicyArn=arn, PolicyDocument=policy_document(statements), SetAsDefault=True + ) + iam.delete_policy_version(PolicyArn=arn, VersionId=old_default) + print(" - synced") + + managed_arns.add(arn) + return arn + + +def ensure_role( + name: str, + trust_policy: JsonObject, + managed_policies: Sequence[str], + *, + max_session_duration: timedelta = timedelta(hours=1), +) -> str: + iam = boto3.client("iam") + arn = role_arn(name) + trust_json = policy_document([trust_policy]) + + print(f" - role {name}") + with _suppress("EntityAlreadyExists"): + iam.create_role( + RoleName=name, + AssumeRolePolicyDocument=trust_json, + Tags=[{"Key": k, "Value": v} for k, v in TAGS.items()], + ) + + iam.update_assume_role_policy(RoleName=name, PolicyDocument=trust_json) + iam.update_role( + RoleName=name, MaxSessionDuration=int(max_session_duration.total_seconds()) + ) + iam.tag_role(RoleName=name, Tags=[{"Key": k, "Value": v} for k, v in TAGS.items()]) + + attached_resp = iam.list_attached_role_policies(RoleName=name) + attached_arns = {p["PolicyArn"] for p in attached_resp["AttachedPolicies"]} + + desired_arns = set(managed_policies) + for pa in desired_arns - attached_arns: + print(f" - attaching {pa}") + iam.attach_role_policy(RoleName=name, PolicyArn=pa) + for pa in attached_arns - desired_arns: + print(f" - detaching {pa}") + iam.detach_role_policy(RoleName=name, PolicyArn=pa) + print(" - synced") + + managed_arns.add(arn) + return arn + + +def ensure_user( + name: str, + managed_policies: Sequence[str], +) -> str: + iam = boto3.client("iam") + arn = f"arn:aws:iam::{ACCOUNT_ID}:user/{name}" + + print(f" - user {name}") + with _suppress("EntityAlreadyExists"): + iam.create_user( + UserName=name, Tags=[{"Key": k, "Value": v} for k, v in TAGS.items()] + ) + + iam.tag_user(UserName=name, Tags=[{"Key": k, "Value": v} for k, v in TAGS.items()]) + + attached_resp = iam.list_attached_user_policies(UserName=name) + attached_arns = {p["PolicyArn"] for p in attached_resp["AttachedPolicies"]} + + desired_arns = set(managed_policies) + for pa in desired_arns - attached_arns: + print(f" - attaching {pa}") + iam.attach_user_policy(UserName=name, PolicyArn=pa) + for pa in attached_arns - desired_arns: + print(f" - detaching {pa}") + iam.detach_user_policy(UserName=name, PolicyArn=pa) + + for ip_name in iam.list_user_policies(UserName=name)["PolicyNames"]: + print(f" - deleting inline policy {ip_name}") + iam.delete_user_policy(UserName=name, PolicyName=ip_name) + print(" - synced") + + managed_arns.add(arn) + return arn + + +def ensure_security_group( + name: str, + description: str, + *, + region: str, + ingress: Sequence[IpPermission] = (), +) -> str: + ec2 = boto3.client("ec2", region_name=region) + + print(f" - security group {name}") + with _suppress("InvalidGroup.Duplicate"): + ec2.create_security_group( + GroupName=name, + Description=description, + TagSpecifications=[ + { + "ResourceType": "security-group", + "Tags": [{"Key": k, "Value": v} for k, v in TAGS.items()], + } + ], + ) + + desc = ec2.describe_security_groups( + Filters=[{"Name": "group-name", "Values": [name]}], + ) + group_id = desc["SecurityGroups"][0]["GroupId"] + ec2.create_tags( + Resources=[group_id], Tags=[{"Key": k, "Value": v} for k, v in TAGS.items()] + ) + + current_perms = desc["SecurityGroups"][0]["IpPermissions"] + if current_perms: + ec2.revoke_security_group_ingress(GroupId=group_id, IpPermissions=current_perms) + if ingress: + ec2.authorize_security_group_ingress( + GroupId=group_id, IpPermissions=list(ingress) + ) + print(" - synced") + + managed_arns.add(f"arn:aws:ec2:{region}:{ACCOUNT_ID}:security-group/{group_id}") + return group_id + + +def ensure_instance_profile(name: str, role_name: str) -> str: + iam = boto3.client("iam") + + print(f" - instance profile {name}") + with _suppress("EntityAlreadyExists"): + iam.create_instance_profile( + InstanceProfileName=name, + Tags=[{"Key": k, "Value": v} for k, v in TAGS.items()], + ) + + resp = iam.get_instance_profile(InstanceProfileName=name) + existing_roles = {r["RoleName"] for r in resp["InstanceProfile"]["Roles"]} + if role_name not in existing_roles: + print(f" - adding role {role_name}") + iam.add_role_to_instance_profile(InstanceProfileName=name, RoleName=role_name) + for extra in existing_roles - {role_name}: + print(f" - removing role {extra}") + iam.remove_role_from_instance_profile(InstanceProfileName=name, RoleName=extra) + print(" - synced") + + arn = f"arn:aws:iam::{ACCOUNT_ID}:instance-profile/{name}" + managed_arns.add(arn) + return arn + + +def ensure_bucket( + name: str, + region: BucketLocationConstraintType | Literal["us-east-1"], + *, + policy: Sequence[JsonObject], + block_public_acls: bool = True, + ignore_public_acls: bool = True, + block_public_policy: bool = True, + restrict_public_buckets: bool = True, + lifecycle: BucketLifecycleConfigurationTypeDef | None = None, +) -> str: + s3 = boto3.client("s3", region_name=region) + + print(f" - bucket {name} ({region})") + # us-east-1 can't be specified as a LocationConstraint: + # https://github.com/boto/boto3/issues/125 + with _suppress("BucketAlreadyOwnedByYou"): + if region != "us-east-1": + s3.create_bucket( + Bucket=name, CreateBucketConfiguration={"LocationConstraint": region} + ) + else: + s3.create_bucket(Bucket=name) + + # Tags + s3.put_bucket_tagging( + Bucket=name, + Tagging={"TagSet": [{"Key": k, "Value": v} for k, v in TAGS.items()]}, + ) + + # Ownership controls (disable ACLs) + s3.put_bucket_ownership_controls( + Bucket=name, + OwnershipControls={"Rules": [{"ObjectOwnership": "BucketOwnerEnforced"}]}, + ) + + # Public access block + s3.put_public_access_block( + Bucket=name, + PublicAccessBlockConfiguration={ + "BlockPublicAcls": block_public_acls, + "IgnorePublicAcls": ignore_public_acls, + "BlockPublicPolicy": block_public_policy, + "RestrictPublicBuckets": restrict_public_buckets, + }, + ) + + # Bucket policy + s3.put_bucket_policy( + Bucket=name, + Policy=policy_document(policy), + ) + + # Lifecycle + if lifecycle is not None: + s3.put_bucket_lifecycle_configuration( + Bucket=name, + LifecycleConfiguration=lifecycle, + ) + else: + s3.delete_bucket_lifecycle(Bucket=name) + print(" - synced") + + arn = f"arn:aws:s3:::{name}" + managed_arns.add(arn) + return arn + + +def ensure_launch_template( + name: str, + *, + region: str, + image_id: str, + instance_type: InstanceTypeType, + security_group_ids: Sequence[str], + user_data: str, + iam_instance_profile: str, + instance_name: str, +) -> str: + ec2 = boto3.client("ec2", region_name=region) + + instance_tags = {**TAGS, "Name": instance_name} + template_data: RequestLaunchTemplateDataTypeDef = { + "ImageId": image_id, + "InstanceType": instance_type, + "SecurityGroupIds": list(security_group_ids), + "UserData": base64.b64encode(user_data.encode()).decode(), + "IamInstanceProfile": {"Name": iam_instance_profile}, + "TagSpecifications": [ + { + "ResourceType": "instance", + "Tags": [{"Key": k, "Value": v} for k, v in instance_tags.items()], + }, + { + "ResourceType": "volume", + "Tags": [{"Key": k, "Value": v} for k, v in instance_tags.items()], + }, + ], + } + + print(f" - launch template {name}") + with _suppress("InvalidLaunchTemplateName.AlreadyExistsException"): + ec2.create_launch_template( + LaunchTemplateName=name, + LaunchTemplateData=template_data, + TagSpecifications=[ + { + "ResourceType": "launch-template", + "Tags": [{"Key": k, "Value": v} for k, v in TAGS.items()], + } + ], + ) + + # AWS caps launch template versions; delete non-defaults to make room + # before adding ours, then clean up the old default afterwards. + versions = ec2.describe_launch_template_versions(LaunchTemplateName=name)[ + "LaunchTemplateVersions" + ] + old_default = next(str(v["VersionNumber"]) for v in versions if v["DefaultVersion"]) + non_default = [str(v["VersionNumber"]) for v in versions if not v["DefaultVersion"]] + if non_default: + ec2.delete_launch_template_versions( + LaunchTemplateName=name, + Versions=non_default, + ) + + version_resp = ec2.create_launch_template_version( + LaunchTemplateName=name, + LaunchTemplateData=template_data, + ) + version = version_resp["LaunchTemplateVersion"]["VersionNumber"] + ec2.modify_launch_template(LaunchTemplateName=name, DefaultVersion=str(version)) + ec2.delete_launch_template_versions( + LaunchTemplateName=name, + Versions=[old_default], + ) + + lt_id = ec2.describe_launch_templates( + LaunchTemplateNames=[name], + )["LaunchTemplates"][0]["LaunchTemplateId"] + + print(f" - version {version}") + + managed_arns.add(f"arn:aws:ec2:{region}:{ACCOUNT_ID}:launch-template/{lt_id}") + return lt_id + + +def ensure_auto_scaling_group( + name: str, + *, + region: str, + launch_template: str, + min_size: int, + max_size: int, + desired_capacity: int, +) -> str: + autoscaling = boto3.client("autoscaling", region_name=region) + ec2 = boto3.client("ec2", region_name=region) + + subnets = ec2.describe_subnets( + Filters=[{"Name": "default-for-az", "Values": ["true"]}], + )["Subnets"] + vpc_zone_id = ",".join(s["SubnetId"] for s in subnets) + launch_template_spec: LaunchTemplateSpecificationTypeDef = { + "LaunchTemplateId": launch_template, + "Version": "$Default", + } + + tags: list[ASGTag] = [ + { + "ResourceId": name, + "ResourceType": "auto-scaling-group", + "Key": k, + "Value": v, + "PropagateAtLaunch": False, + } + for k, v in TAGS.items() + ] + + print(f" - auto scaling group {name}") + with _suppress("AlreadyExists"): + autoscaling.create_auto_scaling_group( + AutoScalingGroupName=name, + LaunchTemplate=launch_template_spec, + MinSize=min_size, + MaxSize=max_size, + DesiredCapacity=desired_capacity, + VPCZoneIdentifier=vpc_zone_id, + Tags=tags, + ) + + autoscaling.update_auto_scaling_group( + AutoScalingGroupName=name, + LaunchTemplate=launch_template_spec, + MinSize=min_size, + MaxSize=max_size, + DesiredCapacity=desired_capacity, + VPCZoneIdentifier=vpc_zone_id, + ) + autoscaling.create_or_update_tags(Tags=tags) + print(" - synced") + + resp = autoscaling.describe_auto_scaling_groups(AutoScalingGroupNames=[name]) + asg = resp["AutoScalingGroups"][0] + managed_arns.add(asg["AutoScalingGroupARN"]) + instance_ids = [i["InstanceId"] for i in asg["Instances"]] + running = _collect_instance_arns(instance_ids) + if running: + print(" - running resources:") + for arn, instance_name in sorted(running.items()): + print(f" - {arn}: {instance_name}") + managed_arns.add(arn) + + return name + + +def ensure_parameter( + name: str, value: str, param_type: ParameterTypeType = "String" +) -> None: + ssm = boto3.client("ssm", region_name=CI_RUNNER_REGION) + + print(f" - {name} ({param_type}, {len(value)} bytes)") + ssm.put_parameter(Name=name, Value=value, Type=param_type, Overwrite=True) + managed_arns.add(f"arn:aws:ssm:{CI_RUNNER_REGION}:{ACCOUNT_ID}:parameter{name}") + + +def _collect_instance_arns(instance_ids: Sequence[str]) -> dict[str, str]: + """Describe instances and collect their associated ARNs. + + Returns a dict mapping instance/volume/ENI ARNs to the instance's Name tag. + """ + if not instance_ids: + return {} + + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + result: dict[str, str] = {} + arn_prefix = f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}" + + resp = ec2.describe_instances(InstanceIds=list(instance_ids)) + for reservation in resp["Reservations"]: + for instance in reservation["Instances"]: + tags = {t["Key"]: t["Value"] for t in instance.get("Tags", ())} + name = tags.get("Name", instance["InstanceId"]) + + result[f"{arn_prefix}:instance/{instance['InstanceId']}"] = name + for ni in instance.get("NetworkInterfaces", ()): + result[f"{arn_prefix}:network-interface/{ni['NetworkInterfaceId']}"] = ( + name + ) + for bdm in instance.get("BlockDeviceMappings", ()): + if vol_id := bdm.get("Ebs", {}).get("VolumeId"): + result[f"{arn_prefix}:volume/{vol_id}"] = name + + return result + + +def check_unmanaged() -> set[str]: + print("\n## Resource audit") + + explorer = boto3.client("resource-explorer-2") + resources = { + r["Arn"] + for page in explorer.get_paginator("search").paginate( + QueryString=RESOURCES_QUERY + ) + for r in page["Resources"] + } + + if managed := resources & managed_arns: + print(f" - {len(managed)} managed resources") + resources -= managed + + if known_unmanaged := resources & set(UNMANAGED_RESOURCES): + print("\n### Known unmanaged") + for arn in sorted(known_unmanaged): + print(f" - {arn}: {UNMANAGED_RESOURCES[arn]}") + resources -= known_unmanaged + + runners = { + arn: name + for arn, name in _collect_instance_arns([ + arn.rsplit("/", 1)[-1] + for arn in resources + if fnmatch.fnmatch(arn, "arn:aws:ec2:*:instance/*") + ]).items() + if name.startswith(RUNNER_NAME_PREFIX) + } + if runners: + print("\n### Running CI jobs") + for arn in sorted(runners): + print(f" - {arn}: {runners[arn]}") + resources -= set(runners) + + if resources: + print("\n### Unexpected resources") + for arn in sorted(resources): + print(f" - {RED}{arn}{RESET}") + + return resources diff --git a/lib/aws/infra_definitions.py b/lib/aws/infra_definitions.py new file mode 100644 index 0000000000..78003cd05d --- /dev/null +++ b/lib/aws/infra_definitions.py @@ -0,0 +1,506 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +"""Desired-state definitions for cockpit CI AWS infrastructure. + +Defines sync_* functions that create-or-update IAM policies, roles, users, +instance profiles, S3 buckets, EC2 resources, and SSM parameters from the +constants in account.py. Safe to run repeatedly — existing resources are +updated in place, nothing is deleted. +""" + +import json +from collections.abc import Mapping, Sequence +from datetime import timedelta +from pathlib import Path + +from ..aio.jsonutil import JsonObject +from .account import ( + ACCOUNT_ID, + CI_IMAGES_BUCKETS, + CI_RUNNER_REGION, + DISPATCHER_ASG, + DISPATCHER_NAME, + DISPATCHER_PARAMS, + DISPATCHER_ROLE, + IMAGE_DOWNLOAD_ROLE, + IMAGE_UPLOAD_ROLE, + LOGS_BUCKET, + LOGS_REGION, + LOGS_WRITE_ROLE, + REDHAT_SSO_IMAGE_DOWNLOAD_MAX_SESSION, + REDHAT_SSO_IMAGE_DOWNLOAD_ROLE, + REDHAT_SSO_SAML_PROVIDER_ARN, + RUNNER_INSTANCE_SLUG_TAG, + RUNNER_NAME_PREFIX, + SSH_SECURITY_GROUP, + TAGS, +) +from .ensure_resource import ( + check_unmanaged, + ensure_auto_scaling_group, + ensure_bucket, + ensure_instance_profile, + ensure_launch_template, + ensure_parameter, + ensure_policy, + ensure_role, + ensure_security_group, + ensure_user, +) + + +def allow( + action: str | Sequence[str], + resource: str | Sequence[str], + condition: JsonObject | None = None, +) -> JsonObject: + return { + "Effect": "Allow", + "Action": action, + "Resource": resource, + **({"Condition": condition} if condition is not None else {}), + } + + +def trust( + action: str, + principal: JsonObject, + condition: JsonObject | None = None, +) -> JsonObject: + return { + "Effect": "Allow", + "Principal": principal, + "Action": action, + **({"Condition": condition} if condition is not None else {}), + } + + +def sync_iam() -> None: + print("\n## IAM") + via_sts_assume_role = trust( + "sts:AssumeRole", {"AWS": f"arn:aws:iam::{ACCOUNT_ID}:root"} + ) + + # Managed policies + + # Download any image (mostly useful for RHEL as the others are public) + policy_images_download = ensure_policy( + "cockpit-ci-images-download", + [ + allow( + "s3:GetObject", + [f"arn:aws:s3:::{name}/*" for name in CI_IMAGES_BUCKETS], + ), + ], + ) + + # Upload, enumerate, and prune images + policy_images_upload = ensure_policy( + "cockpit-ci-images-upload", + [ + allow( + # image-refresh calls image-prune which needs to enumerate the bucket + "s3:ListBucket", + [f"arn:aws:s3:::{name}" for name in CI_IMAGES_BUCKETS], + ), + allow( + [ + "s3:PutObject", # actually upload images + "s3:PutObjectAcl", # TODO: maybe remove this? + "s3:DeleteObject", # image-prune + ], + [f"arn:aws:s3:::{name}/*" for name in CI_IMAGES_BUCKETS], + ), + ], + ) + + # Write to the logs bucket + policy_logs_write = ensure_policy( + "cockpit-ci-logs-write", + [ + allow( + ["s3:PutObject", "s3:DeleteObject"], + f"arn:aws:s3:::{LOGS_BUCKET}/*", + ), + ], + ) + + # Dispatch CI jobs: EC2 lifecycle, S3 logs, STS, SSM, KMS + policy_dispatcher = ensure_policy( + "cockpit-ci-dispatcher", + [ + # RunInstances requires separate statements: one for + allow( + # Allow creating runner instances only if they are tagged + # according to corporate policy, are named as we expect, + # and are tagged with cockpit-ci-slug. + "ec2:RunInstances", + [ + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:instance/*", + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:volume/*", + ], + condition={ + "StringEquals": { + f"aws:RequestTag/{key}": value for key, value in TAGS.items() + }, + "StringLike": { + "aws:RequestTag/Name": f"{RUNNER_NAME_PREFIX}*", + f"aws:RequestTag/{RUNNER_INSTANCE_SLUG_TAG}": "*", + }, + }, + ), + allow( + # We have two separate goals here + "ec2:RunInstances", + [ + # These three are consumed + f"arn:aws:ec2:{CI_RUNNER_REGION}::image/*", + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:security-group/*", + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:subnet/*", + # + # We create this, but can't add tags to it. See + # https://github.com/aws/aws-cli/issues/2865 + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:network-interface/*", + ], + ), + allow( + # We need to have this permission to write the tags on + # instances and volumes. + "ec2:CreateTags", + [ + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:instance/*", + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:volume/*", + ], + condition={ + # But restricted to being done via our RunInstances + "StringEquals": {"ec2:CreateAction": "RunInstances"} + }, + ), + # + # The dispatcher terminates instances that have been running too long + allow( + "ec2:TerminateInstances", + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:instance/*", + condition={ + "StringLike": {f"aws:ResourceTag/{RUNNER_INSTANCE_SLUG_TAG}": "*"}, + }, + ), + # + # Read-only information required by the dispatcher + allow( + [ + "ec2:DescribeInstances", + "ec2:DescribeImages", + "ec2:DescribeSecurityGroups", + ], + "*", + ), + # + # Used by the dispatcher for dashboard.html + allow("s3:PutObject", f"arn:aws:s3:::{LOGS_BUCKET}/*"), + # + # This is how the dispatcher mints scoped tokens for the runners + allow( + "sts:AssumeRole", + [ + ensure_role( + IMAGE_DOWNLOAD_ROLE, + via_sts_assume_role, + managed_policies=[policy_images_download], + max_session_duration=timedelta(hours=6), + ), + ensure_role( + IMAGE_UPLOAD_ROLE, + via_sts_assume_role, + managed_policies=[policy_images_download, policy_images_upload], + max_session_duration=timedelta(hours=6), + ), + ensure_role( + LOGS_WRITE_ROLE, + via_sts_assume_role, + managed_policies=[policy_logs_write], + max_session_duration=timedelta(hours=6), + ), + ], + ), + # + # This is how the dispatcher gains access to SSM parameters + # (amqp server address) and secrets (github-token, amqp tls + # client certificate, etc.) We also need to authorize kms to + # decrypt SecureString secrets for us, but only if it's invoked + # via SSM. + allow( + ["ssm:GetParameter", "ssm:GetParametersByPath"], + f"arn:aws:ssm:{CI_RUNNER_REGION}:{ACCOUNT_ID}:parameter{DISPATCHER_PARAMS}*", + ), + allow( + "kms:Decrypt", + f"arn:aws:kms:{CI_RUNNER_REGION}:{ACCOUNT_ID}:alias/aws/ssm", + condition={ + "StringEquals": { + "kms:ViaService": f"ssm.{CI_RUNNER_REGION}.amazonaws.com", + }, + "StringLike": { + "kms:EncryptionContext:PARAMETER_ARN": + # + f"arn:aws:ssm:{CI_RUNNER_REGION}:{ACCOUNT_ID}:parameter{DISPATCHER_PARAMS}/*", + }, + }, + ), + ], + ) + + # Manual operations via infractl + policy_infractl = ensure_policy( + "cockpit-ci-infractl", + [ + allow( + [ + "ec2:DescribeInstances", + "autoscaling:DescribeAutoScalingGroups", + ], + "*", + ), + allow( + ["ec2:TerminateInstances", "ec2:GetConsoleOutput"], + f"arn:aws:ec2:{CI_RUNNER_REGION}:{ACCOUNT_ID}:instance/*", + condition={ + "StringLike": {"aws:ResourceTag/Name": "cockpit-ci/*"}, + }, + ), + allow( + "autoscaling:SetDesiredCapacity", + f"arn:aws:autoscaling:{CI_RUNNER_REGION}:{ACCOUNT_ID}:autoScalingGroup:*:autoScalingGroupName/{DISPATCHER_ASG}", + ), + allow( + "ssm:PutParameter", + [ + f"arn:aws:ssm:{CI_RUNNER_REGION}:{ACCOUNT_ID}:parameter/cockpit-ci/dispatcher-url", + f"arn:aws:ssm:{CI_RUNNER_REGION}:{ACCOUNT_ID}:parameter{DISPATCHER_PARAMS}/runner-url", + ], + ), + ], + ) + + # Roles + ensure_role( + DISPATCHER_ROLE, + trust( + "sts:AssumeRole", {"Service": "ec2.amazonaws.com"} + ), # acquire via EC2 instance profile + managed_policies=[policy_dispatcher], + ) + ensure_role( + REDHAT_SSO_IMAGE_DOWNLOAD_ROLE, + trust( # acquire via SAML federation from RedHat SSO + "sts:AssumeRoleWithSAML", + {"Federated": REDHAT_SSO_SAML_PROVIDER_ARN}, + condition={ + "StringEquals": {"SAML:aud": "https://signin.aws.amazon.com/saml"}, + }, + ), + managed_policies=[policy_images_download], + max_session_duration=REDHAT_SSO_IMAGE_DOWNLOAD_MAX_SESSION, + ) + + # Users + ensure_user( + # This is a testing account. It can do anything that the others can. + "cockpit-ci", + managed_policies=[ + policy_dispatcher, + policy_images_download, + policy_images_upload, + policy_logs_write, + ], + ) + ensure_user( + "cockpit-ci-infractl", + managed_policies=[policy_infractl], + ) + + # Instance profiles + ensure_instance_profile(DISPATCHER_ROLE, DISPATCHER_ROLE) + + +def sync_s3() -> None: + print("\n## S3") + # Images buckets — public read except RHEL images + for name, region in CI_IMAGES_BUCKETS.items(): + ensure_bucket( + name, + region, + block_public_policy=False, + restrict_public_buckets=False, + policy=[ + { + "Effect": "Allow", + "Principal": "*", + "Action": "s3:GetObject", + "NotResource": f"arn:aws:s3:::{name}/rhel*", + } + ], + ) + + # Logs bucket — public read on everything, expire after 90 days + ensure_bucket( + LOGS_BUCKET, + LOGS_REGION, + block_public_policy=False, + restrict_public_buckets=False, + policy=[ + { + "Effect": "Allow", + "Principal": "*", + "Action": "s3:GetObject", + "Resource": f"arn:aws:s3:::{LOGS_BUCKET}/*", + } + ], + lifecycle={ + "Rules": [ + { + "ID": "expire-after-90-days", + "Status": "Enabled", + "Filter": {"Prefix": ""}, + "Expiration": {"Days": 90}, + } + ], + }, + ) + + +def sync_ec2() -> None: + print("\n## EC2") + sg_id = ensure_security_group( + SSH_SECURITY_GROUP, + "SSH access for cockpit CI instances", + region=CI_RUNNER_REGION, + ingress=[ + { + "IpProtocol": "tcp", + "FromPort": 22, + "ToPort": 22, + "IpRanges": [{"CidrIp": "0.0.0.0/0"}], + } + ], + ) + + ssh_authorized_keys = Path(__file__).parent / "authorized_keys" + launch_template_id = ensure_launch_template( + "cockpit-ci-dispatcher", + region=CI_RUNNER_REGION, + instance_name=DISPATCHER_NAME, + image_id="resolve:ssm:/aws/service/debian/release/trixie/latest/amd64", + instance_type="t3.medium", + security_group_ids=[sg_id], + iam_instance_profile=DISPATCHER_ROLE, + user_data="#cloud-config\n" + + json.dumps({ + "ssh_authorized_keys": ssh_authorized_keys.read_text().splitlines(), + "packages": [ + "git", + "python3-boto3", + "python3-httpx", + "python3-pika", + "python3-yarl", + ], + "write_files": [ + # { + # 'path': '/usr/local/bin/poweroff-stale-instance', + # 'permissions': '0755', + # 'content': r"""#!/bin/sh + # awk '{exit !($1 < 86400)}' /proc/uptime || poweroff -f + # """, + # }, + { + "path": "/usr/local/bin/dispatcher", + "permissions": "0755", + "content": r"""#!/bin/bash + set -euxo pipefail + + bots_url="${1-$( + aws ssm get-parameter \ + --name /cockpit-ci/dispatcher-url \ + --query Parameter.Value --output text + )}" + + mkdir "${RUNTIME_DIRECTORY}/bots" + cd "${RUNTIME_DIRECTORY}/bots" + + curl -sSLf "${bots_url}" | tar xz --strip-components=1 + exec python3 -m lib.aws.dispatcher + """, + }, + { + "path": "/usr/local/lib/systemd/system/dispatcher.service", + "content": r""" + [Unit] + Description=Cockpit CI Dispatcher + Wants=network-online.target + After=network-online.target + # FailureAction=poweroff-immediate + # StartLimitAction=poweroff-immediate + StartLimitBurst=3 + StartLimitIntervalSec=60 + + [Service] + Type=exec + DynamicUser=yes + RuntimeDirectory=dispatcher + WorkingDirectory=/run/dispatcher + ExecStart=/usr/local/bin/dispatcher + # ExecStopPost=+/usr/local/bin/poweroff-stale-instance + Restart=on-failure + RestartSec=30 + RestartPreventExitStatus=5 + """, + }, + ], + "runcmd": [ + ["systemctl", "daemon-reload"], + ["systemctl", "enable", "--now", "dispatcher.service"], + ], + }), + ) + + ensure_auto_scaling_group( + DISPATCHER_ASG, + region=CI_RUNNER_REGION, + min_size=0, + max_size=1, + desired_capacity=1, # see infractl dispatcher up/down + launch_template=launch_template_id, + ) + + +def update_bots_urls( + cockpit_bots_url: str, *, dispatcher: bool = True, runner: bool = True +) -> None: + if dispatcher: + ensure_parameter("/cockpit-ci/dispatcher-url", cockpit_bots_url) + if runner: + ensure_parameter(f"{DISPATCHER_PARAMS}/runner-url", cockpit_bots_url) + + +def sync_ssm(*, cockpit_bots_url: str, secrets: Mapping[str, str]) -> None: + print("\n## SSM") + ensure_parameter( + f"{DISPATCHER_PARAMS}/amqp-server", + "amqp-cockpit.apps.ocp.cloud.ci.centos.org:443", + ) + update_bots_urls(cockpit_bots_url) + ensure_parameter(f"{DISPATCHER_PARAMS}/max-active", "50") + ensure_parameter(f"{DISPATCHER_PARAMS}/max-awaiting-logs", "20") + for name, value in sorted(secrets.items()): + ensure_parameter( + f"{DISPATCHER_PARAMS}/{name}", value, param_type="SecureString" + ) + + +def sync_infra(*, cockpit_bots_url: str, secrets: Mapping[str, str]) -> set[str]: + sync_iam() + sync_s3() + sync_ec2() + sync_ssm(cockpit_bots_url=cockpit_bots_url, secrets=secrets) + return check_unmanaged() diff --git a/lib/aws/infractl.py b/lib/aws/infractl.py new file mode 100644 index 0000000000..afbbc453e6 --- /dev/null +++ b/lib/aws/infractl.py @@ -0,0 +1,333 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +"""Cockpit CI AWS infrastructure management.""" + +import argparse +import logging +import os +import sys +from collections.abc import Iterator +from pathlib import Path +from typing import TYPE_CHECKING + +import boto3 + +from ..aio.jsonutil import get_str +from ..github import GitHub +from .account import CI_RUNNER_REGION, DISPATCHER_ASG, DISPATCHER_NAME +from .ec2 import ( + describe_runner_instances, + get_instance_ip, + get_instance_slug, + get_instance_state, +) +from .infra_definitions import sync_infra, update_bots_urls + +if TYPE_CHECKING: + from types_boto3_ec2.type_defs import InstanceTypeDef + +logger = logging.getLogger(__name__) + + +def resolve_bots_ref(ref: str) -> str: + api = GitHub(repo="cockpit-project/bots") + result = api.get_obj(f"commits/{ref}", None) + if result is None: + raise SystemExit(f"ref {ref!r} not found on github.com/cockpit-project/bots") + + sha = get_str(result, "sha") + logger.info("bots ref %r → %s", ref, sha) + return sha + + +def ssh_to_instance(instance: "InstanceTypeDef", user: str) -> None: + ip = get_instance_ip(instance) + if not ip: + sys.exit(f"instance {instance['InstanceId']} has no public IP") + logger.debug("exec ssh %s@%s", user, ip) + cmd = ["ssh", "-Fnone", "-oKnownHostsCommand=/bin/echo %H %t %K", f"{user}@{ip}"] + os.execvp(cmd[0], cmd) + + +# --- dispatcher --- + + +def get_dispatcher_instances() -> Iterator["InstanceTypeDef"]: + logger.debug("looking up instance %r", DISPATCHER_NAME) + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + resp = ec2.describe_instances( + Filters=[{"Name": "tag:Name", "Values": [DISPATCHER_NAME]}], + ) + for reservation in resp["Reservations"]: + for instance in reservation["Instances"]: + logger.debug("found instance %r", instance["InstanceId"]) + yield instance + + +def get_dispatcher_instance() -> InstanceTypeDef | None: + for instance in get_dispatcher_instances(): + if get_instance_state(instance) == "running": + return instance + return None + + +def dispatcher_up() -> None: + autoscaling = boto3.client("autoscaling", region_name=CI_RUNNER_REGION) + logger.debug("setting desired capacity to 1 for %r", DISPATCHER_NAME) + autoscaling.set_desired_capacity( + AutoScalingGroupName=DISPATCHER_ASG, + DesiredCapacity=1, + ) + print("desired capacity set to 1") + + +def dispatcher_down() -> None: + autoscaling = boto3.client("autoscaling", region_name=CI_RUNNER_REGION) + logger.debug("setting desired capacity to 0 for %r", DISPATCHER_NAME) + autoscaling.set_desired_capacity( + AutoScalingGroupName=DISPATCHER_ASG, + DesiredCapacity=0, + ) + print("desired capacity set to 0") + + +def dispatcher_restart() -> None: + autoscaling = boto3.client("autoscaling", region_name=CI_RUNNER_REGION) + resp = autoscaling.describe_auto_scaling_groups( + AutoScalingGroupNames=[DISPATCHER_ASG] + ) + instance_ids = [ + inst["InstanceId"] + for asg in resp["AutoScalingGroups"] + for inst in asg["Instances"] + ] + if not instance_ids: + sys.exit("no instances in ASG") + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + logger.debug("terminating %r", instance_ids) + ec2.terminate_instances(InstanceIds=instance_ids) + print(f"detached and terminated {instance_ids}, ASG will launch a replacement") + + +def dispatcher_ssh() -> None: + instance = get_dispatcher_instance() + if instance is None: + sys.exit("no running dispatcher instance found") + ssh_to_instance(instance, "admin") + + +def dispatcher_update(args: argparse.Namespace) -> None: + sha = resolve_bots_ref(args.bots_ref) + bots_url = f"https://github.com/cockpit-project/bots/archive/{sha}.tar.gz" + logger.debug("updating SSM with bots URL %r", bots_url) + print(f"bots ref: {args.bots_ref}") + print(f"bots sha: {sha}") + update_bots_urls( + bots_url, + dispatcher=not args.only_runner, + runner=not args.only_dispatcher, + ) + + +def dispatcher_status() -> None: + autoscaling = boto3.client("autoscaling", region_name=CI_RUNNER_REGION) + resp = autoscaling.describe_auto_scaling_groups( + AutoScalingGroupNames=[DISPATCHER_ASG], + ) + for asg in resp["AutoScalingGroups"]: + print( + f"{asg['AutoScalingGroupName']} desired: {asg['DesiredCapacity']} " + f"min: {asg['MinSize']} " + f"max: {asg['MaxSize']}" + ) + for inst in asg["Instances"]: + print(f" {inst['InstanceId']} {inst['LifecycleState']}") + + for instance in get_dispatcher_instances(): + print( + instance["InstanceId"], + f"state: {get_instance_state(instance)}", + f"ip: {get_instance_ip(instance)}", + ) + + +# --- runner --- + + +def runner_list(*, show_all: bool = False) -> None: + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + for instance in describe_runner_instances(ec2): + state = get_instance_state(instance) + if not show_all and state == "terminated": + continue + print(f" {get_instance_slug(instance)} {instance['InstanceId']} {state}") + + +def runner_terminate(slug: str) -> None: + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + instances = describe_runner_instances(ec2, slug=slug) + if not instances: + sys.exit(f"no runner instances found for slug {slug!r}") + instance_ids = [inst["InstanceId"] for inst in instances] + logger.debug("terminating %r", instance_ids) + ec2.terminate_instances(InstanceIds=instance_ids) + print(f"terminated {instance_ids}") + + +def runner_ssh(slug: str) -> None: + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + running = [ + inst + for inst in describe_runner_instances(ec2, slug=slug) + if get_instance_state(inst) == "running" + ] + if not running: + sys.exit(f"no running runner instance found for slug {slug!r}") + ssh_to_instance(running[0], "core") + + +def runner_console(slug: str) -> None: + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + instances = describe_runner_instances(ec2, slug=slug) + if not instances: + sys.exit(f"no runner instance found for slug {slug!r}") + instance_id = instances[0]["InstanceId"] + logger.debug("getting console output for %r", instance_id) + resp = ec2.get_console_output(InstanceId=instance_id) + output = resp.get("Output", "") + if output: + print(output, end="") + else: + print("(no console output available yet)") + print("\nNote: console log output is delayed by ~10 minutes", file=sys.stderr) + + +# --- main --- + + +def main() -> None: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--debug", "-d", action="store_true") + sub = parser.add_subparsers(dest="command", required=True) + + sync_parser = sub.add_parser( + "sync", help="Sync AWS infrastructure to desired state" + ) + sync_parser.add_argument( + "--bots-ref", required=True, help="bots ref to deploy" + ) + sync_parser.add_argument( + "--secrets-dir", + type=Path, + default=None, + help="directory containing secret files to upload to SSM", + ) + + disp = sub.add_parser("dispatcher", help="Manage the dispatcher instance") + disp_sub = disp.add_subparsers(dest="dispatcher_command", required=True) + disp_sub.add_parser("up", help="Start the dispatcher instance") + disp_sub.add_parser("down", help="Stop the dispatcher instance") + disp_sub.add_parser( + "restart", help="Terminate and let the ASG replace the instance" + ) + disp_sub.add_parser("ssh", help="SSH to the dispatcher instance") + disp_sub.add_parser("status", help="Show dispatcher instance status") + update_parser = disp_sub.add_parser( + "update", help="Update bots SHA in SSM parameters" + ) + update_parser.add_argument( + "--bots-ref", required=True, help="bots ref to deploy" + ) + update_group = update_parser.add_mutually_exclusive_group() + update_group.add_argument( + "--only-dispatcher", action="store_true", help="only update the dispatcher URL" + ) + update_group.add_argument( + "--only-runner", action="store_true", help="only update the runner URL" + ) + + runner = sub.add_parser("runner", help="Manage CI runner instances") + runner_sub = runner.add_subparsers(dest="runner_command", required=True) + + runner_list_parser = runner_sub.add_parser("list", help="List CI runner instances") + runner_list_parser.add_argument( + "-a", "--show-all", action="store_true", help="Include terminated instances" + ) + + runner_terminate_parser = runner_sub.add_parser( + "terminate", help="Terminate CI runner instances by slug" + ) + runner_terminate_parser.add_argument("slug") + + runner_ssh_parser = runner_sub.add_parser( + "ssh", help="SSH to a CI runner instance by slug" + ) + runner_ssh_parser.add_argument("slug") + + runner_console_parser = runner_sub.add_parser( + "console", help="Show EC2 console output for a runner" + ) + runner_console_parser.add_argument("slug") + + args = parser.parse_args() + + logging.basicConfig( + level=logging.DEBUG if args.debug else logging.WARNING, + format="%(name)s: %(message)s", + ) + + match args.command: + case "sync": + sha = resolve_bots_ref(args.bots_ref) + bots_url = f"https://github.com/cockpit-project/bots/archive/{sha}.tar.gz" + secrets = ( + {p.name: p.read_text() for p in args.secrets_dir.iterdir()} + if args.secrets_dir + else {} + ) + + print("\n# Deployment") + print(f" - bots ref: {args.bots_ref}") + print(f" - bots sha: {sha}") + print(f" - bots url: {bots_url}") + print(f" - secrets: {list(secrets)}") + + unexpected = sync_infra(cockpit_bots_url=bots_url, secrets=secrets) + + if not args.secrets_dir and any(":parameter/" in arn for arn in unexpected): + print("\nHint: some unexpected SSM parameters were found.") + print( + "Use --secrets-dir to provide secret files if they need updating." + ) + print(f"\nRun `{parser.prog} dispatcher restart` to pick up changes.") + case "dispatcher": + match args.dispatcher_command: + case "up": + dispatcher_up() + case "down": + dispatcher_down() + case "restart": + dispatcher_restart() + case "ssh": + dispatcher_ssh() + case "status": + dispatcher_status() + case "update": + dispatcher_update(args) + case "runner": + match args.runner_command: + case "list": + runner_list(show_all=args.show_all) + case "terminate": + runner_terminate(args.slug) + case "ssh": + runner_ssh(args.slug) + case "console": + runner_console(args.slug) + + +if __name__ == "__main__": + main() diff --git a/lib/aws/jobconfig.py b/lib/aws/jobconfig.py new file mode 100644 index 0000000000..b0bb175c57 --- /dev/null +++ b/lib/aws/jobconfig.py @@ -0,0 +1,182 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import json +import logging +from collections.abc import Iterator, Mapping, Sequence +from datetime import timedelta +from typing import TYPE_CHECKING + +from ..aio.jsonutil import JsonObject, JsonValue +from ..s3 import S3Key +from .account import ( + ACCOUNT_ID, + IMAGE_DOWNLOAD_ROLE, + IMAGE_UPLOAD_ROLE, + LOGS_BUCKET, + LOGS_URL, + LOGS_WRITE_ROLE, +) + +if TYPE_CHECKING: + from types_boto3_sts import STSClient + +logger = logging.getLogger(__name__) + + +def assume_role( + sts: STSClient, + name: str, + policy: JsonObject | None = None, + duration: timedelta = timedelta(hours=1), +) -> S3Key: + logger.debug("assuming role %r", name) + if policy is not None: + response = sts.assume_role( + RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/{name}", + RoleSessionName=name, + Policy=json.dumps(policy), + DurationSeconds=int(duration.total_seconds()), + ) + else: + response = sts.assume_role( + RoleArn=f"arn:aws:iam::{ACCOUNT_ID}:role/{name}", + RoleSessionName=name, + DurationSeconds=int(duration.total_seconds()), + ) + creds = response["Credentials"] + logger.debug("got credentials expiring %s", creds["Expiration"]) + return S3Key(creds["AccessKeyId"], creds["SecretAccessKey"], creds["SessionToken"]) + + +def provide_secrets( + sts: STSClient, + secrets: Sequence[str], + params: Mapping[str, str] = {}, + duration: timedelta = timedelta(hours=1), +) -> Iterator[tuple[str, JsonValue]]: + + logger.debug("providing secrets %r with duration %r", secrets, duration) + + if "image-upload" in secrets: + yield ( + "image-upload", + str(assume_role(sts, IMAGE_UPLOAD_ROLE, duration=duration)), + ) + + if "image-download" in secrets: + yield ( + "image-download", + str(assume_role(sts, IMAGE_DOWNLOAD_ROLE, duration=duration)), + ) + + if "github-token" in secrets: + yield "github-token", params["github-token"] + + if "fedora-wiki" in secrets: + yield "fedora-wiki", params["fedora-wiki"] + + if "fedora-wiki-staging" in secrets: + yield "fedora-wiki-staging", params["fedora-wiki-staging"] + + +def job_runner_config( + slug: str, + sts: STSClient, + *, + secrets: Sequence[str] = (), + params: Mapping[str, str], + post: bool = False, + credential_duration: timedelta, +) -> JsonObject: + logger.debug( + "building job-runner config for %r (credential_duration=%r)", + slug, + credential_duration, + ) + return { + "container": { + "run-args": [ + # don't run as actual root + # "--userns=auto", + # TODO: need to either share the userns or netns, otherwise + # multicast UDP won't work (which is how multiple VMs talk to + # each other). Let's use --network=host for now. + # "--network=host", + # general resource limits + "--device=/dev/kvm", + "--memory=56g", + "--pids-limit=16384", + "--shm-size=1024m", + # /tmp on tmpfs + "--tmpfs=/tmp:size=32g", + "--env=TEST_OVERLAY_DIR=/tmp", + # identity + "--env=GIT_COMMITTER_NAME=Cockpituous", + "--env=GIT_COMMITTER_EMAIL=cockpituous@cockpit-project.org", + "--env=GIT_AUTHOR_NAME=Cockpituous", + "--env=GIT_AUTHOR_EMAIL=cockpituous@cockpit-project.org", + ], + "secrets": { + "github-token": [ + "--env=COCKPIT_GITHUB_TOKEN_FILE=/run/secrets/github-token", + "--volume=%{github-token}:/run/secrets/github-token:ro,Z,U", + ], + "image-download": [ + "--env=COCKPIT_S3_KEY_DIR=/run/secrets/s3", + "--volume=%{image-download}:/run/secrets/s3/amazonaws.com:ro,Z,U", + ], + "image-upload": [ + "--env=COCKPIT_S3_KEY_DIR=/run/secrets/s3", + "--volume=%{image-upload}:/run/secrets/s3/amazonaws.com:ro,Z,U", + ], + "fedora-wiki": [ + "--volume=%{fedora-wiki}:/run/secrets/fedora-wiki.json:ro,Z,U", + "--env=COCKPIT_FEDORA_WIKI_TOKEN=/run/secrets/fedora-wiki.json", + ], + "fedora-wiki-staging": [ + "--volume=%{fedora-wiki-staging}:/run/secrets/fedora-wiki-staging.json:ro,Z,U", + "--env=COCKPIT_FEDORA_WIKI_STAGING_TOKEN=/run/secrets/fedora-wiki-staging.json", + ], + }, + }, + "forge": { + "github": { + "post": post, + "token": params["github-token"], + } + }, + "logs": { + "attach-journal": True, + "driver": "s3", + "s3": { + "url": LOGS_URL, + "acl": "", + "user-agent": "job-runner (cockpit-project/bots)", + "key": str( + assume_role( + sts, + LOGS_WRITE_ROLE, + duration=credential_duration, + policy={ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": ["s3:PutObject", "s3:DeleteObject"], + "Resource": f"arn:aws:s3:::{LOGS_BUCKET}/{slug}/*", + } + ], + }, + ) + ), + }, + }, + "secrets": { + "inline": dict( + provide_secrets(sts, secrets, params, duration=credential_duration) + ), + }, + } diff --git a/lib/aws/launch_runner.py b/lib/aws/launch_runner.py new file mode 100644 index 0000000000..fc53dd6028 --- /dev/null +++ b/lib/aws/launch_runner.py @@ -0,0 +1,183 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import annotations + +import argparse +import contextlib +import json +import logging +import shlex +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from typing import TYPE_CHECKING + +import boto3 +import botocore.exceptions + +from ..aio.jsonutil import typechecked +from .account import CI_RUNNER_REGION, DISPATCHER_PARAMS +from .dispatcher import load_parameters, prepare_and_launch +from .ec2 import get_instance_ip + +if TYPE_CHECKING: + from types_boto3_ec2 import EC2Client + from types_boto3_ec2.literals import InstanceStateNameType + from types_boto3_ec2.type_defs import InstanceTypeDef as Instance + +logger = logging.getLogger(__name__) + + +def watch_instance( + ec2: EC2Client, + instance_id: str, + *, + start: float = 0, + wait_for_state: InstanceStateNameType = "terminated", +) -> Instance: + prev_state: InstanceStateNameType | None = None + while True: + try: + response = ec2.describe_instances(InstanceIds=[instance_id]) + info = response["Reservations"][0]["Instances"][0] + state = info["State"]["Name"] + except botocore.exceptions.ClientError as exc: + # https://docs.aws.amazon.com/ec2/latest/devguide/eventual-consistency.html + if exc.response["Error"]["Code"] != "InvalidInstanceID.NotFound": + raise + if time.clock_gettime(time.CLOCK_BOOTTIME) - start < 30: + time.sleep(2.5) + continue + + sys.exit(f"instance {instance_id} not found after 30s") + + if state in ("terminated", "stopped") and wait_for_state != state: + sys.exit( + f"instance {instance_id} reached {state} waiting for {wait_for_state}" + ) + + if info and state != prev_state: + parts: list[str] = [state] + if ip := get_instance_ip(info): + parts.append(f"public={ip}") + if private_ip := info.get("PrivateIpAddress"): + parts.append(f"private={private_ip}") + print(" ".join(parts)) + prev_state = state + + if info and state == wait_for_state: + return info + + time.sleep(2.5) + + +def wait_for_ssh(info: Instance) -> str: + dns = info.get("PublicDnsName", "") + if not dns: + sys.exit(f"instance {info['InstanceId']} has no PublicDnsName") + while True: + try: + with socket.create_connection((dns, 22), timeout=1): + return dns + except OSError: + time.sleep(1) + + +def terminate_and_wait(ec2: EC2Client, instance_id: str) -> None: + ec2.terminate_instances(InstanceIds=[instance_id]) + watch_instance(ec2, instance_id) + + +def main() -> None: + parser = argparse.ArgumentParser() + # fmt: off + parser.add_argument("--debug", action="store_true", help="Enable debug logging") + parser.add_argument("--ami", help="FCOS AMI ID (default: latest)") + parser.add_argument("--instance-type", default="m8id.4xlarge") + parser.add_argument("--ssh-key", type=Path, + help="Additional SSH public key file to authorize for the core user") + parser.add_argument("--parameters", default=f"ssm:{DISPATCHER_PARAMS}/", + help="Parameter source: ssm:PREFIX, json:DATA, or dir:PATH") + parser.add_argument("--param", action="append", default=[], + help="Override a parameter: --param key=value") + parser.add_argument("job_json", nargs="?", + help="Job specification as JSON string (default: shell)") + # fmt: on + args = parser.parse_args() + + if args.job_json is None: + args.job_json = json.dumps({ + "slug": "shell", + "repo": "cockpit-project/bots", + "command": ["/usr/bin/sleep", "1h", "50m"], + }) + + logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO) + + ssh_keys: list[str] = [] + if args.ssh_key: + ssh_key_text = args.ssh_key.read_text() + if "PRIVATE KEY" in ssh_key_text: + sys.exit("No.") + ssh_keys.extend(ssh_key_text.strip().splitlines()) + + ec2 = boto3.client("ec2", region_name=CI_RUNNER_REGION) + sts = boto3.client("sts", region_name=CI_RUNNER_REGION) + + params = load_parameters(args.parameters) + for override in args.param: + key, _, value = override.partition("=") + logger.debug("overriding parameter %r=%r", key, value) + params[key] = value + + job = typechecked(json.loads(args.job_json), dict) + + with contextlib.ExitStack() as stack: + key_dir = stack.enter_context(tempfile.TemporaryDirectory()) + key_path = f"{key_dir}/id" + subprocess.run( + ["ssh-keygen", "-t", "ed25519", "-N", "", "-f", key_path], + check=True, + stdout=subprocess.DEVNULL, + ) + ssh_keys.append(Path(f"{key_path}.pub").read_text().strip()) + + instance_id = prepare_and_launch( + ec2, + sts, + job=job, + params=params, + bots_url=params["runner-url"], + instance_type=args.instance_type, + post=False, + ami=args.ami, + ssh_keys=ssh_keys, + ) + print(f"launched {instance_id}") + stack.callback(terminate_and_wait, ec2, instance_id) + + info = watch_instance( + ec2, + instance_id, + start=time.clock_gettime(time.CLOCK_BOOTTIME), + wait_for_state="running", + ) + print("waiting for ssh...") + dns = wait_for_ssh(info) + ssh_cmd = [ + "ssh", + "-Fnone", + "-oKnownHostsCommand=/bin/echo %H %t %K", + f"-i{key_path}", + f"core@{dns}", + ] + print(f"\nInstance is online and accessible:\n {shlex.join(ssh_cmd)}\n") + subprocess.run(ssh_cmd) + + +if __name__ == "__main__": + main() diff --git a/lib/aws/ruff.toml b/lib/aws/ruff.toml new file mode 100644 index 0000000000..e0a2e59529 --- /dev/null +++ b/lib/aws/ruff.toml @@ -0,0 +1,10 @@ +# The code in lib/aws is meant to work with Python 3.13 and uses the +# default options for formatting (88 character lines, double quotes). + +extend = "../../pyproject.toml" +target-version = "py313" + +# All of these are defaults +lint.ignore = [] +line-length = 88 +format.quote-style = "double" diff --git a/lib/gssapi_saml_sts.py b/lib/gssapi_saml_sts.py index 531176a6d9..730e69947c 100644 --- a/lib/gssapi_saml_sts.py +++ b/lib/gssapi_saml_sts.py @@ -21,6 +21,13 @@ from datetime import datetime, timedelta, timezone from pathlib import Path +from lib.aws.account import ( + ACCOUNT_ID, + REDHAT_SSO_IDP_URL, + REDHAT_SSO_IMAGE_DOWNLOAD_MAX_SESSION, + REDHAT_SSO_IMAGE_DOWNLOAD_ROLE, + REDHAT_SSO_SAML_PROVIDER_ARN, +) from lib.directories import xdg_cache_home from lib.s3 import S3Key @@ -87,11 +94,14 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None return result[0] -def aws_sts_assume_role(role_arn: str, provider_arn: str, saml_assertion: str) -> str: +def aws_sts_assume_role( + account_id: str, role: str, provider_arn: str, saml_assertion: str, duration: timedelta +) -> str: """Exchange a SAML assertion for temporary AWS credentials via STS. Returns the raw XML response body. """ + role_arn = f'arn:aws:iam::{account_id}:role/{role}' logger.debug('assuming role %r via STS', role_arn) params = urllib.parse.urlencode({ @@ -100,7 +110,7 @@ def aws_sts_assume_role(role_arn: str, provider_arn: str, saml_assertion: str) - 'RoleArn': role_arn, 'PrincipalArn': provider_arn, 'SAMLAssertion': saml_assertion, - 'DurationSeconds': 43200, # 12 hours + 'DurationSeconds': int(duration.total_seconds()), }).encode() request = urllib.request.Request(_STS_URL, data=params, method='POST') @@ -180,16 +190,20 @@ def load_from_cache(role_name: str) -> S3Key | None: class SAMLTarget: idp_url: str provider_arn: str - role_arn: str + account_id: str + role: str + max_session_duration: timedelta # Red Hat employee IdP (SAML via Kerberos) → AWS IAM role for cockpit-ci-images # Rover group: https://rover.redhat.com/groups/group/it-cloud-aws-727920394381-cockpit-ci-images-download TARGETS = { 'https://cockpit-ci-images*.s3.*.amazonaws.com/rhel-*': SAMLTarget( - idp_url='https://auth.redhat.com/auth/realms/EmployeeIDP/protocol/saml/clients/itaws', - provider_arn='arn:aws:iam::727920394381:saml-provider/RedHatInternal', - role_arn='arn:aws:iam::727920394381:role/727920394381-cockpit-ci-images-download', + idp_url=REDHAT_SSO_IDP_URL, + provider_arn=REDHAT_SSO_SAML_PROVIDER_ARN, + account_id=ACCOUNT_ID, + role=REDHAT_SSO_IMAGE_DOWNLOAD_ROLE, + max_session_duration=REDHAT_SSO_IMAGE_DOWNLOAD_MAX_SESSION, ), } @@ -207,14 +221,12 @@ def try_key(url: str, use_cache: bool = True) -> S3Key | None: if target is None: return None - role_name = target.role_arn.rsplit('/', 1)[-1] - if use_cache: - if key := load_from_cache(role_name): + if key := load_from_cache(target.role): return key saml = _get_saml_assertion(target.idp_url) - xml = aws_sts_assume_role(target.role_arn, target.provider_arn, saml) + xml = aws_sts_assume_role(target.account_id, target.role, target.provider_arn, saml, target.max_session_duration) - key, _expiration = save_to_cache(role_name, xml) + key, _expiration = save_to_cache(target.role, xml) return key diff --git a/lib/html/dashboard/dashboard.html b/lib/html/dashboard/dashboard.html new file mode 100644 index 0000000000..6ea1459191 --- /dev/null +++ b/lib/html/dashboard/dashboard.html @@ -0,0 +1,49 @@ + + + + + + cockpit CI + + + + + + + diff --git a/lib/html/dashboard/dashboard.js b/lib/html/dashboard/dashboard.js new file mode 100644 index 0000000000..bb4056d82f --- /dev/null +++ b/lib/html/dashboard/dashboard.js @@ -0,0 +1,228 @@ +// @ts-check + +/** + * @typedef {{ + * state: string, + * ip: string | null, + * launch_time: string, + * }} Instance + * + * @typedef {{ + * observed_instances?: string[], + * launched_instance?: string, + * human?: string, + * logs_visible?: boolean, + * }} Job + * + * @typedef {{ + * jobs: Record, + * instances: Record, + * }} DashboardData + */ + +import { css, html, LitElement } from "https://cdn.jsdelivr.net/gh/lit/dist@3/all/lit-all.min.js"; + +class CiDashboard extends LitElement { + /** @override */ + static properties = { + data: { state: true }, + error: { state: true }, + _now: { state: true }, + _showTerminated: { state: true }, + }; + + /** @override */ + static styles = css` + :host { display: block; } + table { border-collapse: collapse; width: 100%; max-width: 900px; } + th, td { padding: 0.4rem 0.8rem; text-align: left; } + th { color: var(--cyan); font-weight: bold; font-size: 0.85rem; text-transform: uppercase; border-bottom: 2px solid var(--border); } + + .job-row td { border-top: 1px solid var(--border); } + .slug { font-weight: bold; } + .slug a { text-decoration: none; } + .slug a:hover { text-decoration: underline; } + .state-queued { color: var(--dim); font-style: italic; } + + .instance-row td { font-size: 0.85rem; color: var(--dim); background: var(--instance-bg); } + .instance-row td:first-child { padding-left: 2rem; } + .instance-row.ours td:first-child { border-left: 2px solid var(--blue); padding-left: calc(2rem - 2px); } + .instance-id { font-size: 0.8rem; } + .ip { color: var(--cyan); } + .age { text-align: right; } + .state-running { color: var(--green); } + .state-pending { color: var(--yellow); } + .state-terminated, .state-shutting-down { color: var(--red); } + + .error { color: var(--red); } + h1 { color: var(--fg); font-size: 1.1rem; margin-bottom: 1rem; } + .controls { display: flex; align-items: center; gap: 0.5rem; margin-bottom: 1rem; font-size: 0.85rem; color: var(--dim); } + .controls label { cursor: pointer; display: flex; align-items: center; gap: 0.4rem; } + .meta { color: var(--dim); font-size: 0.8rem; margin-top: 1rem; } + `; + + constructor() { + super(); + /** @type {DashboardData | null} */ + this.data = null; + /** @type {string | null} */ + this.error = null; + /** @type {number} */ + this._now = Date.now(); + /** @type {boolean} */ + this._showTerminated = false; + /** @type {number | undefined} */ + this._fetchInterval = undefined; + /** @type {number | undefined} */ + this._tickInterval = undefined; + } + + /** @override @returns {void} */ + connectedCallback() { + super.connectedCallback(); + this._fetch(); + this._fetchInterval = setInterval(() => this._fetch(), 5000); + this._tickInterval = setInterval(() => { + this._now = Date.now(); + }, 1000); + } + + /** @override @returns {void} */ + disconnectedCallback() { + super.disconnectedCallback(); + clearInterval(this._fetchInterval); + clearInterval(this._tickInterval); + } + + /** @returns {Promise} */ + async _fetch() { + try { + const resp = await fetch("summary.json"); + if (!resp.ok) { + throw new Error(`${resp.status}`); + } + this.data = await resp.json(); + this.error = null; + } catch (e) { + this.error = /** @type {Error} */ (e).message; + } + } + + /** @override */ + render() { + if (!this.data) { + return this.error ? html`

${this.error}

` : html`

loading...

`; + } + const data = this.data; + const now = this._now; + + /** + * @param {string} iso_string + * @returns {string} + */ + function format_age(iso_string) { + const ms = now - new Date(iso_string).getTime(); + const mins = Math.floor(ms / 60000); + if (mins < 60) { + return `${mins}m`; + } + return `${Math.floor(mins / 60)}h${mins % 60}m`; + } + + /** @param {Job} job */ + function job_instances(job) { + return (job.observed_instances || []).filter((iid) => data.instances[iid]); + } + + /** + * @param {string} slug + * @param {Job} job + */ + function render_job(slug, job) { + const instances = job_instances(job).sort((a, b) => + data.instances[a].launch_time.localeCompare(data.instances[b].launch_time), + ); + + const label = job.human || slug; + const slugCell = job.logs_visible ? html`${label}` : label; + + return html` + + ${slugCell} + + ${ + instances.length + ? instances.map((iid) => { + const inst = data.instances[iid]; + return html` + + ${iid} + ${inst.state} + ${inst.ip || ""} + ${format_age(inst.launch_time)} + + `; + }) + : html` + + + queued + + + + ` + } + `; + } + + /** @param {Job} job */ + function is_job_terminated(job) { + const iids = job_instances(job); + return ( + iids.length > 0 && + iids.every((iid) => ["terminated", "shutting-down"].includes(data.instances[iid].state)) + ); + } + + /** @param {Job} job */ + function newest_launch(job) { + const iids = job_instances(job); + if (!iids.length) { + return Infinity; + } + return Math.max(...iids.map((iid) => new Date(data.instances[iid].launch_time).getTime())); + } + + return html` +

cockpit CI

+ ${this.error ? html`

fetch error: ${this.error}

` : ""} +
+ +
+ + + + + + + + + + + ${Object.entries(data.jobs) + .filter(([, job]) => this._showTerminated || !is_job_terminated(job)) + .sort(([, a], [, b]) => newest_launch(b) - newest_launch(a)) + .map(([slug, job]) => render_job(slug, job))} + +
jobstateipage
+ `; + } +} + +customElements.define("ci-dashboard", CiDashboard); diff --git a/lib/html/tsconfig.json b/lib/html/tsconfig.json new file mode 100644 index 0000000000..bf0fde3cee --- /dev/null +++ b/lib/html/tsconfig.json @@ -0,0 +1,13 @@ +{ + "compilerOptions": { + "checkJs": true, + "strict": true, + "noEmit": true, + "noImplicitOverride": true, + "target": "ES2022", + "module": "ES2022", + "moduleResolution": "bundler" + }, + "include": ["*/*.js"], + "files": ["types/lit-cdn.d.ts"] +} diff --git a/lib/html/types/lit-cdn.d.ts b/lib/html/types/lit-cdn.d.ts new file mode 100644 index 0000000000..e5b18b9365 --- /dev/null +++ b/lib/html/types/lit-cdn.d.ts @@ -0,0 +1,27 @@ +declare module "https://cdn.jsdelivr.net/gh/lit/dist@3/all/lit-all.min.js" { + type PropertyDeclaration = { + type?: typeof String | typeof Number | typeof Boolean | typeof Array | typeof Object; + state?: boolean; + }; + + export type CSSResult = { cssText: string }; + + export class LitElement extends HTMLElement { + static properties: Record; + static styles: CSSResult | CSSResult[]; + connectedCallback(): void; + disconnectedCallback(): void; + createRenderRoot(): HTMLElement | DocumentFragment; + requestUpdate(): void; + render(): unknown; + } + + export function css(strings: TemplateStringsArray, ...values: unknown[]): CSSResult; + export function html(strings: TemplateStringsArray, ...values: unknown[]): unknown; + export const nothing: symbol; + export function repeat( + items: Iterable, + keyFn: (item: T) => unknown, + template: (item: T) => unknown, + ): unknown; +} diff --git a/lib/s3.py b/lib/s3.py index fd063b71bf..1fa949b4a1 100644 --- a/lib/s3.py +++ b/lib/s3.py @@ -39,6 +39,9 @@ class S3Key(NamedTuple): secret: str token: str | None = None + def __str__(self) -> str: + return ' '.join(v for v in self if v is not None) + SHA256_NIL = hashlib.sha256(b'').hexdigest() diff --git a/lib/stores.py b/lib/stores.py index fe38ff6e13..55d292d4ea 100644 --- a/lib/stores.py +++ b/lib/stores.py @@ -17,12 +17,12 @@ from collections.abc import Sequence +from lib.aws.account import CI_IMAGES_BUCKETS, LOGS_URL from lib.directories import xdg_config_home # hosted on the public internet, requires a private token for some/all images IMAGE_STORES: Sequence[str] = ( - "https://cockpit-ci-images-fra.s3.eu-central-1.amazonaws.com/", - "https://cockpit-ci-images.s3.us-east-1.amazonaws.com/", + *(f'https://{name}.s3.{region}.amazonaws.com/' for name, region in CI_IMAGES_BUCKETS.items()), ) # locally configured stores in ~/.config/cockpit-dev/image-stores or $COCKPIT_IMAGE_STORES_FILE @@ -35,4 +35,4 @@ LOCAL_STORES: Sequence[str] = data.splitlines() -LOG_STORE = "https://cockpit-ci-logs.s3.us-east-1.amazonaws.com/" +LOG_STORE = LOGS_URL diff --git a/lib/stubs/pika/__init__.pyi b/lib/stubs/pika/__init__.pyi new file mode 100644 index 0000000000..d56681b2b8 --- /dev/null +++ b/lib/stubs/pika/__init__.pyi @@ -0,0 +1,33 @@ +import ssl +from typing import Any + +from .channel import Channel +from .credentials import ExternalCredentials, PlainCredentials +from .spec import Basic + +class SSLOptions: + def __init__(self, context: ssl.SSLContext, *, server_hostname: str = ...) -> None: ... + +class ConnectionParameters: + def __init__( + self, + host: str = ..., + port: int = ..., + *, + ssl_options: SSLOptions = ..., + credentials: PlainCredentials | ExternalCredentials = ..., + ) -> None: ... + +class BasicProperties: + def __init__(self, *, priority: int = ..., **kwargs: Any) -> None: ... + +class BlockingChannel(Channel): + def basic_get( + self, queue: str, auto_ack: bool = ... + ) -> tuple[Basic.GetOk, BasicProperties, bytes] | tuple[None, None, None]: ... + +class BlockingConnection: + def __init__(self, parameters: ConnectionParameters = ...) -> None: ... + def channel(self) -> BlockingChannel: ... + def close(self) -> None: ... + def process_data_events(self, time_limit: float = ...) -> None: ... diff --git a/lib/stubs/pika/adapters/__init__.pyi b/lib/stubs/pika/adapters/__init__.pyi new file mode 100644 index 0000000000..e69de29bb2 diff --git a/lib/stubs/pika/adapters/asyncio_connection.pyi b/lib/stubs/pika/adapters/asyncio_connection.pyi new file mode 100644 index 0000000000..aed0ad7037 --- /dev/null +++ b/lib/stubs/pika/adapters/asyncio_connection.pyi @@ -0,0 +1,18 @@ +from collections.abc import Callable + +from pika import ConnectionParameters +from pika.channel import Channel + +class AsyncioConnection: + def __init__( + self, + parameters: ConnectionParameters | None = ..., + *, + on_open_callback: Callable[[AsyncioConnection], object] | None = ..., + on_open_error_callback: Callable[[AsyncioConnection, Exception], object] | None = ..., + on_close_callback: Callable[[AsyncioConnection, Exception], object] | None = ..., + ) -> None: ... + + def channel(self, *, on_open_callback: Callable[[Channel], object]) -> None: ... + + def close(self) -> None: ... diff --git a/lib/stubs/pika/channel.pyi b/lib/stubs/pika/channel.pyi new file mode 100644 index 0000000000..7b35cc1a94 --- /dev/null +++ b/lib/stubs/pika/channel.pyi @@ -0,0 +1,51 @@ +from collections.abc import Callable +from typing import Any + +from pika import BasicProperties +from pika.frame import Method +from pika.spec import Basic + +class Channel: + def queue_declare( + self, + queue: str, + *, + durable: bool = ..., + passive: bool = ..., + arguments: dict[str, Any] | None = ..., + ) -> Method: ... + + def basic_publish( + self, + exchange: str, + routing_key: str, + body: str | bytes, + *, + properties: BasicProperties = ..., + ) -> None: ... + + def basic_qos( + self, + *, + prefetch_size: int = ..., + prefetch_count: int = ..., + global_qos: bool = ..., + callback: Callable[..., object] | None = ..., + ) -> None: ... + + def basic_consume( + self, + queue: str, + *, + on_message_callback: Callable[[Channel, Basic.Deliver, BasicProperties | None, bytes], object] = ..., + auto_ack: bool = ..., + arguments: dict[str, Any] | None = ..., + ) -> str: ... + + def basic_cancel(self, consumer_tag: str) -> None: ... + + def basic_ack(self, delivery_tag: int) -> None: ... + + def basic_reject(self, delivery_tag: int, *, requeue: bool = ...) -> None: ... + + def add_on_close_callback(self, callback: Callable[[Channel, Exception], object]) -> None: ... diff --git a/lib/stubs/pika/credentials.pyi b/lib/stubs/pika/credentials.pyi new file mode 100644 index 0000000000..20449e73f3 --- /dev/null +++ b/lib/stubs/pika/credentials.pyi @@ -0,0 +1,5 @@ +class PlainCredentials: + def __init__(self, username: str, password: str) -> None: ... + +class ExternalCredentials: + def __init__(self) -> None: ... diff --git a/lib/stubs/pika/exceptions.pyi b/lib/stubs/pika/exceptions.pyi new file mode 100644 index 0000000000..63b41262ec --- /dev/null +++ b/lib/stubs/pika/exceptions.pyi @@ -0,0 +1,6 @@ +class AMQPError(Exception): ... +class AMQPConnectionError(AMQPError): ... +class ChannelClosedByBroker(AMQPError): + reply_code: int + reply_text: str + def __init__(self, reply_code: int, reply_text: str) -> None: ... diff --git a/lib/stubs/pika/frame.pyi b/lib/stubs/pika/frame.pyi new file mode 100644 index 0000000000..624d2284fe --- /dev/null +++ b/lib/stubs/pika/frame.pyi @@ -0,0 +1,4 @@ +from pika.spec import Queue + +class Method: + method: Queue.DeclareOk diff --git a/lib/stubs/pika/spec.pyi b/lib/stubs/pika/spec.pyi new file mode 100644 index 0000000000..4b457a97ce --- /dev/null +++ b/lib/stubs/pika/spec.pyi @@ -0,0 +1,10 @@ +class Basic: + class Deliver: + delivery_tag: int + + class GetOk: + delivery_tag: int + +class Queue: + class DeclareOk: + message_count: int | None diff --git a/pyproject.toml b/pyproject.toml index ac1d1bee8b..14775cf0b0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,20 +2,32 @@ name = "bots" requires-python = ">= 3.14" +[tool.pyright] +stubPath = "lib/stubs" + [tool.mypy] python_version = "3.14" strict = true follow_imports = 'silent' # https://github.com/python-lsp/pylsp-mypy/issues/81 scripts_are_modules = true # allow checking all scripts in one invocation +mypy_path = 'lib/stubs' warn_return_any = false [[tool.mypy.overrides]] # things which may be unavailable when running checks module = [ + 'boto3', + 'botocore', + 'botocore.*', 'libvirt', 'libvirt_qemu', 'nacl', - 'pika.*', + 'types_boto3_autoscaling.*', + 'types_boto3_ec2', + 'types_boto3_ec2.*', + 'types_boto3_s3.*', + 'types_boto3_ssm.*', + 'types_boto3_sts', ] ignore_missing_imports = true diff --git a/saml-login b/saml-login index 4b9bc9832d..485ad2220e 100755 --- a/saml-login +++ b/saml-login @@ -11,53 +11,99 @@ for use by image-download. """ import argparse +import json import logging +import shlex import sys +from datetime import timedelta +from typing import Literal import gi # type: ignore [import-untyped] from lib.ansi import GREEN, RESET +from lib.aws.account import ( + ACCOUNT_ID, + REDHAT_SSO_IDP_URL, + REDHAT_SSO_IMAGE_DOWNLOAD_MAX_SESSION, + REDHAT_SSO_IMAGE_DOWNLOAD_ROLE, + REDHAT_SSO_SAML_PROVIDER_ARN, +) from lib.gssapi_saml_sts import ( - TARGETS, AuthError, aws_sts_assume_role, cache_path, + load_from_cache, save_to_cache, ) +from lib.s3 import S3Key logger = logging.getLogger(__name__) +def print_credentials(key: S3Key, output: Literal['awscreds', 'env']) -> None: + match output: + case 'env': + print(f'export AWS_ACCESS_KEY_ID={shlex.quote(key.access)}') + print(f'export AWS_SECRET_ACCESS_KEY={shlex.quote(key.secret)}') + if key.token: + print(f'export AWS_SESSION_TOKEN={shlex.quote(key.token)}') + case 'awscreds': + print(json.dumps({ + 'Version': 1, + 'AccessKeyId': key.access, + 'SecretAccessKey': key.secret, + 'SessionToken': key.token, + })) + + def main() -> None: parser = argparse.ArgumentParser(description='Log in to AWS via SAML using a web browser') + # fmt: off parser.add_argument('-d', '--debug', action='store_true', help='Enable debug logging') + parser.add_argument('role', nargs='?', default=REDHAT_SSO_IMAGE_DOWNLOAD_ROLE, + help='IAM role name to assume (default: %(default)s)') + parser.add_argument('--duration', type=lambda s: timedelta(hours=float(s)), + default=REDHAT_SSO_IMAGE_DOWNLOAD_MAX_SESSION, + help='session duration in hours (default: %(default)s)') + parser.add_argument('--output', choices=['env', 'awscreds'], + help='output credentials in the given format') + # fmt: on args = parser.parse_args() if args.debug: logging.basicConfig(level=logging.DEBUG, format='%(name)s: %(message)s') + cached = load_from_cache(args.role) + if cached is not None: + if args.output: + print_credentials(cached, args.output) + else: + print(f'{GREEN}Cached credentials for {args.role} still valid{RESET}') + return + gi.require_version('Gtk', '4.0') gi.require_version('JavaScriptCore', '6.0') gi.require_version('WebKit', '6.0') from gi.repository import GLib, Gtk, JavaScriptCore, WebKit # type: ignore [import-untyped] - target, = TARGETS.values() # only one for now loop = GLib.MainLoop() def on_saml_message_received(_manager: WebKit.UserContentManager, message: JavaScriptCore.Value) -> None: saml_assertion = message.to_string() logger.debug('captured SAML assertion (%d bytes)', len(saml_assertion)) - xml = aws_sts_assume_role(target.role_arn, target.provider_arn, saml_assertion) + xml = aws_sts_assume_role(ACCOUNT_ID, args.role, REDHAT_SSO_SAML_PROVIDER_ARN, saml_assertion, args.duration) - role_name = target.role_arn.rsplit('/', 1)[-1] try: - _key, expiration = save_to_cache(role_name, xml) + key, expiration = save_to_cache(args.role, xml) except AuthError as exc: sys.exit(str(exc)) - local = expiration.astimezone() - print(f'\n{GREEN}STS credentials saved to {cache_path(role_name)}\nValid until {local:%c}{RESET}\n') + if args.output: + print_credentials(key, args.output) + else: + local = expiration.astimezone() + print(f'\n{GREEN}STS credentials saved to {cache_path(args.role)}\nValid until {local:%c}{RESET}\n') loop.quit() content_manager = WebKit.UserContentManager() @@ -88,7 +134,7 @@ def main() -> None: ) webview = WebKit.WebView(user_content_manager=content_manager) - webview.load_uri(target.idp_url) + webview.load_uri(REDHAT_SSO_IDP_URL) win = Gtk.Window( title='SAML Login', diff --git a/test/generate_test_jobs.py b/test/generate_test_jobs.py new file mode 100644 index 0000000000..fec8e64be6 --- /dev/null +++ b/test/generate_test_jobs.py @@ -0,0 +1,241 @@ +# Copyright (C) 2026 Red Hat, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +import argparse +import json +import logging +import time +from collections.abc import Sequence + +import pika + +from lib import distributed_queue, github, testmap +from lib.aio.jsonutil import get_dict, get_str +from lib.jobqueue import QueueEntry + +logger = logging.getLogger(__name__) + + +def resolve_ref(api: github.GitHub, ref: str) -> tuple[str, str | None]: + """Resolve a ref to (sha, branch). + + ref can be 'pr:NUMBER', a branch name, or a SHA. + For PRs, branch is the base branch. For branches, branch is the ref itself. + For bare SHAs, branch is None. + """ + if ref.startswith('pr:'): + pr_nr = int(ref[3:]) + pull = api.get_obj(f"pulls/{pr_nr}") + sha = get_str(get_dict(pull, "head"), "sha") + branch = get_str(get_dict(pull, "base"), "ref") + logger.info("pr %d → sha %s, base branch %r", pr_nr, sha, branch) + return sha, branch + + result = api.get_obj(f"commits/{ref}", None) + if result is None: + raise SystemExit(f"ref {ref!r} not found on github.com/{api.repo}") + + sha = get_str(result, "sha") + logger.debug("resolved %r to %r", ref, sha) + + branches = testmap.tests_for_project(api.repo) + if ref in branches: + return sha, ref + + return sha, None + + +def filter_contexts( + contexts: Sequence[str], + os_filter: Sequence[str], + scenario_filter: Sequence[str], +) -> Sequence[str]: + result: list[str] = [] + for context in contexts: + # skip cross-repo contexts + if '@' in context: + continue + + image, _, scenario = context.partition('/') + + if os_filter and image not in os_filter: + continue + if scenario_filter and not any(s in scenario for s in scenario_filter): + continue + + result.append(context) + + return result + + +def generate_entries( + repo: str, + sha: str, + branch: str | None, + contexts: Sequence[str], + bots_ref: str | None, + extra_env: Sequence[str], + pull: int | None, +) -> Sequence[QueueEntry]: + entries: list[QueueEntry] = [] + timestamp = time.strftime('%Y%m%d-%H%M%S') + + for context in contexts: + image, _, scenario = context.partition('/') + + slug_suffix = context.replace('/', '-').replace('@', '-') + slug = f"ec2-full-{timestamp}-{repo.replace('/', '-')}-{sha[:8]}-{slug_suffix}" + + env: dict[str, str] = { + "TEST_OS": image, + "TEST_REVISION": sha, + } + if scenario: + env["TEST_SCENARIO"] = scenario + if branch: + env["BASE_BRANCH"] = branch + if bots_ref: + env["COCKPIT_BOTS_REF"] = bots_ref + if pull is not None: + env["TEST_PULL"] = str(pull) + + for item in extra_env: + key, _, value = item.partition('=') + env[key] = value + + secrets: list[str] = ["github-token", "image-download"] + if repo == "rhinstaller/anaconda-webui": + secrets.extend(["fedora-wiki", "fedora-wiki-staging"]) + + entries.append({ + "job": { + "repo": repo, + "sha": sha, + "context": context, + "pull": pull, + "command_subject": None, + "slug": slug, + "env": env, + "secrets": secrets, + }, + "human": f"{context}@{repo}#{sha[:12]}", + }) + + return entries + + +def publish(entries: Sequence[QueueEntry]) -> None: + queue = 'public' + with distributed_queue.DistributedQueue('localhost', [queue]) as dq: + for entry in entries: + context = entry['job']['context'] + priority = distributed_queue.MAX_PRIORITY if '/devel' in context else distributed_queue.BASELINE_PRIORITY + properties = pika.BasicProperties(priority=priority) + dq.channel.basic_publish('', queue, json.dumps(entry), properties=properties) + logger.info("published %s", entry['human']) + logger.info("published %d entries", len(entries)) + + +def collect_entries( + repo: str, + branches: dict[str, Sequence[str]], + ref: str | None, + os_filter: Sequence[str], + scenario_filter: Sequence[str], + bots_ref: str | None, + extra_env: Sequence[str], +) -> Sequence[QueueEntry]: + api = github.GitHub(repo=repo) + entries: list[QueueEntry] = [] + + pull: int | None = None + if ref is not None: + sha, resolved_branch = resolve_ref(api, ref) + if ref.startswith('pr:'): + pull = int(ref[3:]) + + if resolved_branch and resolved_branch in branches: + branch_list = {resolved_branch: branches[resolved_branch]} + elif resolved_branch is None: + # bare SHA: run against default branch contexts + default = testmap.get_default_branch(repo) + if default in branches: + branch_list = {default: branches[default]} + else: + branch_list = {} + else: + logger.warning("branch %r not in testmap for %r", resolved_branch, repo) + branch_list = {} + + for branch, contexts in branch_list.items(): + filtered = filter_contexts(contexts, os_filter, scenario_filter) + entries.extend(generate_entries(repo, sha, branch, filtered, bots_ref, extra_env, pull)) + else: + for branch, contexts in branches.items(): + if branch.startswith('_'): + continue + sha, _ = resolve_ref(api, branch) + filtered = filter_contexts(contexts, os_filter, scenario_filter) + entries.extend(generate_entries(repo, sha, branch, filtered, bots_ref, extra_env, None)) + + return entries + + +def main() -> None: + parser = argparse.ArgumentParser( + description='Generate test jobs from the testmap', + epilog='With no arguments, generates jobs for all repos and branches in the testmap.', + ) + parser.add_argument('repo', nargs='?', + help='repository (e.g. cockpit-project/cockpit)') + parser.add_argument('ref', nargs='?', + help='branch, SHA, or pr:NUMBER') + parser.add_argument('--os', action='append', default=[], dest='os_filter', + help='filter by OS image (repeatable)') + parser.add_argument('--scenario', action='append', default=[], dest='scenario_filter', + help='filter by scenario (repeatable)') + parser.add_argument('--bots-ref', + help='bots ref to set as COCKPIT_BOTS_REF') + parser.add_argument('--env', action='append', default=[], dest='extra_env', + help='extra env var as KEY=VALUE (repeatable)') + parser.add_argument('--amqp', action='store_true', + help='publish to AMQP on localhost instead of printing JSONL') + args = parser.parse_args() + + logging.basicConfig(level=logging.INFO) + + all_entries: list[QueueEntry] = [] + + if args.repo: + branches = dict(testmap.tests_for_project(args.repo)) + if not branches: + raise SystemExit(f"repo {args.repo!r} not found in testmap") + all_entries.extend(collect_entries( + args.repo, branches, args.ref, args.os_filter, args.scenario_filter, args.bots_ref, args.extra_env, + )) + else: + for repo, branch_map in testmap.REPO_BRANCH_CONTEXT.items(): + if ':' in repo: + logger.debug("skipping non-github repo %r", repo) + continue + branches = dict(branch_map) + all_entries.extend(collect_entries( + repo, branches, args.ref, args.os_filter, args.scenario_filter, args.bots_ref, args.extra_env, + )) + + if not all_entries: + raise SystemExit("no matching test contexts found") + + logger.info("%d entries", len(all_entries)) + + if args.amqp: + publish(all_entries) + for entry in all_entries: + print(entry['job']['slug']) + else: + for entry in all_entries: + print(json.dumps(entry)) + + +if __name__ == '__main__': + main() diff --git a/test/run b/test/run index 97fc396462..677cf1ec92 100755 --- a/test/run +++ b/test/run @@ -20,3 +20,11 @@ find_python_files() { find_python_files | xargs -0 ruff check --quiet find_python_files | xargs -0 mypy --no-error-summary pytest -vv + +if command -v biome >/dev/null 2>&1 && [ -f biome.jsonc ]; then + biome check +fi + +if command -v tsc >/dev/null 2>&1; then + tsc -p lib/html/tsconfig.json +fi diff --git a/test/test_aio.py b/test/test_aio.py index 9a45028077..dd6bd78598 100644 --- a/test/test_aio.py +++ b/test/test_aio.py @@ -14,7 +14,7 @@ from lib.aio.base import SubjectSpecification from lib.aio.github import GitHub from lib.aio.jobcontext import JobContext -from lib.aio.jsonutil import JsonObject, JsonValue, json_merge_patch +from lib.aio.jsonutil import JsonError, JsonObject, JsonValue, json_merge_patch from lib.aio.s3 import S3LogDriver from lib.aio.util import LRUCache from lib.s3 import S3Key @@ -327,10 +327,10 @@ async def test_secrets_expansion(tmp_path: Path) -> None: home = Path.home() async with JobContext(config_file) as context: - assert context.secrets_args == { - 's3-keys': (f'--volume={home}/.config/s3-keys:/run/secrets/s3-keys:ro',), - 'github-token': ('--env=GITHUB_TOKEN_FILE=/etc/github-token',), - } + assert context.prepare_secrets(['s3-keys', 'github-token'], tmp_path / 'secrets') == [ + f'--volume={home}/.config/s3-keys:/run/secrets/s3-keys:ro', + '--env=GITHUB_TOKEN_FILE=/etc/github-token', + ] async def test_secrets_undefined_error(tmp_path: Path) -> None: @@ -349,9 +349,9 @@ async def test_secrets_undefined_error(tmp_path: Path) -> None: local.directory = '/tmp/logs' ''') - with pytest.raises(SystemExit, match=r"undefined secret '%\{undefined\}'"): - async with JobContext(config_file): - pass + async with JobContext(config_file) as ctx: + with pytest.raises(LookupError, match='undefined'): + ctx.prepare_secrets(['bad'], tmp_path / 'secrets') async def test_inline_secrets(tmp_path: Path) -> None: @@ -375,22 +375,20 @@ async def test_inline_secrets(tmp_path: Path) -> None: local.directory = '/tmp/logs' ''') + secrets_dir = tmp_path / 'secrets' async with JobContext(config_file) as ctx: + result = ctx.prepare_secrets(['github-token', 's3-keys'], secrets_dir) + # Check github-token - assert ctx.secrets_args['github-token'][0] == '-e=COCKPIT_GITHUB_TOKEN_FILE=/x' - token_path = Path(ctx.secrets_args['github-token'][1].removeprefix('-v=').removesuffix(':/x')) + assert result[0] == '-e=COCKPIT_GITHUB_TOKEN_FILE=/x' + token_path = Path(result[1].removeprefix('-v=').removesuffix(':/x')) assert token_path.read_text() == 'ghp_secret123' # Check s3-keys - assert ctx.secrets_args['s3-keys'][0] == '-e=COCKPIT_S3_KEY_DIR=/y' - s3_path = Path(ctx.secrets_args['s3-keys'][1].removeprefix('-v=').removesuffix(':/y')) + assert result[2] == '-e=COCKPIT_S3_KEY_DIR=/y' + s3_path = Path(result[3].removeprefix('-v=').removesuffix(':/y')) assert (s3_path / 's3.example.com').read_text() == 'ABCD Zx2xPa' - # After context closes, temp files should be cleaned up - assert not token_path.exists() - assert not s3_path.exists() - assert not token_path.parent.exists() - async def test_inline_path_nested_error(tmp_path: Path) -> None: config_file = tmp_path / 'config.toml' @@ -404,19 +402,19 @@ async def test_inline_path_nested_error(tmp_path: Path) -> None: default-image = 'ghcr.io/test:latest' [container.secrets] + mydir = ['--volume=%{mydir}:/mnt'] [logs] driver = 'local' local.directory = '/tmp/logs' ''') - with pytest.raises( - SystemExit, - match=r"attribute 'secrets': attribute 'inline': attribute 'mydir': " - r"attribute 'subdir': invalid filename: '\.\./escape'", - ): - async with JobContext(config_file): - pass + async with JobContext(config_file) as ctx: + with pytest.raises( + JsonError, + match=r"attribute 'mydir': attribute 'subdir': invalid filename: '\.\./escape'", + ): + ctx.prepare_secrets(['mydir'], tmp_path / 'secrets') async def test_inline_path_invalid_type(tmp_path: Path) -> None: @@ -431,17 +429,18 @@ async def test_inline_path_invalid_type(tmp_path: Path) -> None: default-image = 'ghcr.io/test:latest' [container.secrets] + badvalue = ['--env=X=%{badvalue}'] [logs] driver = 'local' local.directory = '/tmp/logs' ''') - with pytest.raises( - SystemExit, match=r"attribute 'secrets': attribute 'inline': attribute 'badvalue': must be string or object" - ): - async with JobContext(config_file): - pass + async with JobContext(config_file) as ctx: + with pytest.raises( + JsonError, match=r"attribute 'badvalue': must be string or object" + ): + ctx.prepare_secrets(['badvalue'], tmp_path / 'secrets') async def test_secrets_conflict_error(tmp_path: Path) -> None: @@ -514,5 +513,5 @@ async def test_serialize_roundtrip(tmp_path: Path, monkeypatch: pytest.MonkeyPat # Load from serialized config (simulating remote execution) monkeypatch.setenv('JOB_RUNNER_CONFIG_JSON', json.dumps(serialized)) async with JobContext() as ctx2: - # Secrets should work the same way - assert ctx2.secrets_args['github-token'][0].endswith('/github-token') + result = ctx2.prepare_secrets(['github-token'], tmp_path / 'secrets2') + assert result[0].endswith('/github-token') diff --git a/test/test_job.py b/test/test_job.py index 5d31f7cf8f..a3c3765600 100644 --- a/test/test_job.py +++ b/test/test_job.py @@ -130,7 +130,7 @@ async def mock_job_context(tmp_path: Path) -> AsyncGenerator[Mock, None]: mock_ctx.default_image = 'registry.fedoraproject.org/fedora:latest' mock_ctx.container_cmd = ['podman'] mock_ctx.container_run_args = ['--pull=newer'] - mock_ctx.secrets_args = {} + mock_ctx.prepare_secrets = Mock(return_value=[]) mock_ctx.resolve_subject = AsyncMock(wraps=forge.resolve_subject) mock_ctx.server = server @@ -272,6 +272,33 @@ async def test_run_job_failure( assert get_str(failure_data, 'description').startswith('Container exited with code 1') assert failure_data['target_url'] == 'http://localhost:9000/test-job/log.html' + @pytest.mark.parametrize('side_effect', [ + LookupError('no container.secrets.github-token entry'), + LookupError('secret %{github-token} is not configured'), + ]) + async def test_run_job_missing_secret( + self, + side_effect: LookupError, + log_streamer_mocks: LogStreamerMocks, + mock_job_context: Mock, + ) -> None: + job = Job({ + 'repo': 'cockpit-project/cockpit', + 'sha': 'abc123', + 'context': 'verify/rhel-9', + 'secrets': ['github-token'], + }) + mock_job_context.prepare_secrets = Mock(side_effect=side_effect) + + await run_job(job, mock_job_context) + + post_calls = mock_job_context.server.drain_queue() + assert len(post_calls) == 2 + failure_path, failure_data = post_calls[1] + assert failure_path == '/repos/cockpit-project/cockpit/statuses/abc123' + assert failure_data['state'] == 'failure' + assert str(side_effect) in get_str(failure_data, 'description') + @patch('lib.aio.job.run_container') async def test_run_job_failure_with_report( self,