Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions hooks/playbooks/config_cluster_for_disconnected_deployment.yml
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
mirror_location: "{{ disconnect_working_dir }}/mirror_location"
local_registry: "{{ disconnect_working_dir }}/local_registry"
oc_mirror_cert_manager_catalog_url: "{{ cifmw_cert_manager_catalog_url | default('registry.redhat.io/redhat/redhat-operator-index:v4.18') }}"
verify_image_signatures: "{{ cifmw_disconnected_verify_image_signatures | default(false) }}"
tasks:
- name: Create disconnected working directories
ansible.builtin.file:
Expand Down Expand Up @@ -295,6 +296,18 @@
retries: 10
delay: 30

- name: Generate ClusterImagePolicies from mirror data
cifmw.general.generate_cluster_image_policies:
input_dir: "{{ mirror_location }}/working-dir/cluster-resources"
output_file: "{{ disconnect_working_dir }}/ClusterImagePolicies.yaml"
when: verify_image_signatures

- name: Apply cluster image policies yaml
kubernetes.core.k8s:
state: present
src: "{{ disconnect_working_dir }}/ClusterImagePolicies.yaml"
when: verify_image_signatures

- name: Wait until the OpenShift cluster is stable
ansible.builtin.command:
cmd: oc adm wait-for-stable-cluster --minimum-stable-period=5s --timeout=30m
139 changes: 139 additions & 0 deletions plugins/modules/generate_cluster_image_policies.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
#!/usr/bin/python

# Copyright: (c) 2026, Red Hat
# GNU General Public License v3.0+ (see COPYING or
# https://www.gnu.org/licenses/gpl-3.0.txt)

DOCUMENTATION = r"""
---
module: generate_cluster_image_policies

short_description: Generates cluster image policy file for image signature verification

description:
- Looks up files generated by oc-mirror v2 and creates a ClusterImagePolicy file containing
each mirror/source pair. Executing oc apply -f on the ClusterImagePolicy file creates the
cluster image policies that may be used for image signature verification when pulling
container image files from oc-mirror.

options:
input_dir:
description:
- Absolute path to cluster-resources mirror directory
required: True
type: str
output_file:
description:
- Absolute path to directory when ClusterImagePolicy file is created
required: True
type: str

author:
- David Rosenfeld (@drosenfe)
"""

EXAMPLES = r"""
- name: Generate ClusterImagePolicies from mirror data
cifmw.general.generate_cluster_image_policies:
input_dir: "{{ mirror_location }}/working-dir/cluster-resources"
output_file: "{{ disconnect_working_dir }}/ClusterImagePolicies.yaml"
"""

import os
import yaml
from ansible.module_utils.basic import AnsibleModule

NO_POLICIES = (
"No cluster image policies were created. Please check image digest mirror set."
)


def main():
mod_args = {
"input_dir": {"required": True, "type": "str"},
"output_file": {"required": True, "type": "str"},
}

module = AnsibleModule(argument_spec=mod_args, supports_check_mode=False)

result = {
"success": False,
"changed": False,
"debug": {},
}

# Directory where oc-mirror YAML files are stored
input_dir = module.params.get("input_dir")
output_file = module.params.get("output_file")

# Red Hat public key: security.access.redhat.com/data/63405576.txt (between BEGIN and END, passed through | base64 -w0)
sigstore_key = """LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlJQ0lqQU5CZ2txaGtpRzl3MEJBUUVGQUFPQ0FnOEFNSUlDQ2dLQ0FnRUEwQVN5dUgyVExXdkJVcVBIWjRJcAo3NWc3RW5jQmtnUUhkSm5qenhBVzVLUVRNaC9zaUJvQi9Cb1NydGlQTXduQ2hiVENuUU9JUWVadURpRm5odUo3Ck0vRDNiN0pvWDBtMTIzTmNDU242N21BZGpCYTZCZzZrdWtaZ0NQNFpVWmVFU2FqV1gvRWp5bEZjUkZPWFc1N3AKUkRDRU40MkovallsVnF0K2c5K0dya2VyOFN6ODZIM2wwdGJxT2RqYnovVnhIWWh3RjBjdFVNSHN5VlJEcTJRUAp0cXpOWGxtbE1oUy9Qb0ZyNlI0dS83SENuL0srTGVnY08yZkFGT2I0MEt2S1NLS1ZENmxld1VaRXJob3AxQ2dKClhqRHRHbW1POWRHTUY3MW1mNkhFZmFLU2R5K0VFNmlTRjJBMlZ2OVFoQmF3TWlxMmtPekVpTGc0bkFkSlQ4d2cKWnJNQW1QQ3FHSXNYTkdaNC9RK1lUd3dsY2UzZ2xxYjVMOXRmTm96RWRTUjlOODVERVNmUUxRRWRZM0NhbHdLTQpCVDFPRWhFWDF3SFJDVTRkck1PZWo2Qk5XMFZ0c2NHdEhtQ3JzNzRqUGV6aHdOVDh5cGt5UytUMHpUNFRzeTZmClZYa0o4WVNIeWVuU3pNQjJPcDJidnNFM2dyWStzNzRXaEc5VUlBNkRCeGNUaWUxNU5Tekt3Znphb05XT0RjTEYKcDdCWThhYUhFMk1xRnhZRlgrSWJqcGtRUmZhZVFRc291REZkQ2tYRUZWZlBwYkQyZGs2RmxlYU1UUHV5eHRJVApnalZFdEdRSzJxR0NGR2lRSEZkNGhmVitlQ0E2M0pybzF6MHpvQk01QmJJSVEzK2VWRnd0M0FsWnA1VVZ3cjZkCnNlY3FraS95cm12M1kwZHFaOVZPbjNVQ0F3RUFBUT09Ci0tLS0tRU5EIFBVQkxJQyBLRVktLS0tLQ=="""

# Collect mirror-source pairs
pairs = []

for fname in os.listdir(input_dir):
if fname.endswith("oc-mirror.yaml"):
with open(os.path.join(input_dir, fname)) as f:
docs = list(yaml.safe_load_all(f))
for doc in docs:
if not doc or "spec" not in doc:
continue
key = (
"imageDigestMirrors"
if "imageDigestMirrors" in doc["spec"]
else "imageTagMirrors"
)
for entry in doc["spec"].get(key, []):
source = entry["source"]
for mirror in entry["mirrors"]:
pairs.append((mirror, source))

# Deduplicate
unique_pairs = sorted(set(pairs))

# Generate policy YAMLs
policies = []
for mirror, source in unique_pairs:
name = mirror.split("/")[-1].replace(".", "-").replace(":", "-")
policy = {
"apiVersion": "config.openshift.io/v1",
"kind": "ClusterImagePolicy",
"metadata": {"name": f"enforce-signatures-quay-mirror-{name}"},
"spec": {
"scopes": [mirror],
"policy": {
"type": "sigstore",
"rootOfTrust": {
"policyType": "PublicKey",
"publicKey": {"keyData": sigstore_key},
},
"signedIdentity": {
"type": "RemapIdentity",
"matchPolicy": "RemapIdentity",
"remapIdentity": {"prefix": mirror, "signedPrefix": source},
},
},
},
}
policies.append(policy)

# Output to file
with open(output_file, "w") as f:
f.write(
"---\n" + "\n---\n".join(yaml.dump(p, sort_keys=False) for p in policies)
)

print(f"Generated {len(policies)} ClusterImagePolicy objects in {output_file}")

# Ensure some cluster image policies were created
if len(policies) == 0:
error = NO_POLICIES
module.fail_json(msg=error, **result)

result["success"] = True
module.exit_json(**result)


if __name__ == "__main__":
main()
Loading