Security Audit / Infrastructure Scan (IaC) (push) Successful in 1m12s
Deploy Infrastructure / plan (push) Successful in 1m30s
Security Audit / Docker Image Scan (docker.io/apache/tika:latest) (push) Failing after 4m55s
Security Audit / Docker Image Scan (docker.io/gotenberg/gotenberg:8) (push) Failing after 5m35s
Deploy Infrastructure / apply (push) Successful in 1m7s
Security Audit / Docker Image Scan (docker.io/library/postgres:15) (push) Failing after 1m7s
Security Audit / Docker Image Scan (docker.io/library/redis:7) (push) Failing after 59s
Deploy Infrastructure / integration-tests (push) Failing after 10s
Security Audit / Docker Image Scan (ghcr.io/icereed/paperless-gpt:latest) (push) Failing after 52s
Security Audit / Docker Image Scan (shenxn/protonmail-bridge:latest) (push) Successful in 2m1s
Security Audit / Docker Image Scan (ghcr.io/paperless-ngx/paperless-ngx:latest) (push) Failing after 3m57s
- Add recursive function to collect resources from child modules - Update resource address patterns for BPG provider (proxmox_virtual_environment_container instead of proxmox_lxc) - Handle tags as arrays instead of comma-separated strings - Use pattern matching for gitea/runner detection 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
import json
|
|
import sys
|
|
import os
|
|
|
|
def get_all_resources(planned_values):
|
|
"""Recursively collect all resources from root_module and child_modules."""
|
|
resources = []
|
|
|
|
root_module = planned_values.get('root_module', {})
|
|
resources.extend(root_module.get('resources', []))
|
|
|
|
# Recursively get resources from child modules
|
|
def collect_from_modules(module):
|
|
for child in module.get('child_modules', []):
|
|
resources.extend(child.get('resources', []))
|
|
collect_from_modules(child)
|
|
|
|
collect_from_modules(root_module)
|
|
return resources
|
|
|
|
def validate_plan(plan_path):
|
|
if not os.path.exists(plan_path):
|
|
print(f"Error: Plan file {plan_path} not found.")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
with open(plan_path, 'r') as f:
|
|
plan = json.load(f)
|
|
except json.JSONDecodeError as e:
|
|
print(f"Error decoding JSON from {plan_path}: {e}")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Error reading file {plan_path}: {e}")
|
|
sys.exit(1)
|
|
|
|
# Get all resources including those in child modules
|
|
planned_values = plan.get('planned_values', {})
|
|
resources = get_all_resources(planned_values)
|
|
|
|
# Check 1: Gitea Server must exist
|
|
# Resource address: module.gitea_srv_renner_uno.proxmox_virtual_environment_container.this
|
|
gitea_exists = False
|
|
for res in resources:
|
|
addr = res.get('address', '')
|
|
if 'gitea' in addr.lower() and 'proxmox_virtual_environment_container' in addr:
|
|
gitea_exists = True
|
|
break
|
|
|
|
if not gitea_exists:
|
|
print("ERROR: Validation Failed! Gitea server is missing in the planned state (it might be deleted).")
|
|
sys.exit(1)
|
|
|
|
# Check 2: At least one runner must remain
|
|
# Runner address pattern: module.runners["905"].proxmox_virtual_environment_container.this
|
|
runner_count = 0
|
|
for res in resources:
|
|
addr = res.get('address', '')
|
|
if 'runners[' in addr and 'proxmox_virtual_environment_container' in addr:
|
|
runner_count += 1
|
|
|
|
if runner_count < 1:
|
|
print("ERROR: Validation Failed! No Gitea runners found in the planned state.")
|
|
sys.exit(1)
|
|
|
|
# Check 3: Prevent recreation of non-transient nodes
|
|
# We fail if any node that doesnt have the transient tag would be recreated
|
|
resource_changes = plan.get('resource_changes', [])
|
|
for change in resource_changes:
|
|
actions = change.get('change', {}).get('actions', [])
|
|
# Recreation typically involves both delete and create actions for the same resource instance
|
|
if 'delete' in actions and 'create' in actions:
|
|
address = change.get('address', 'unknown')
|
|
after = change.get('change', {}).get('after', {})
|
|
|
|
# If after is None, we can't check tags, but recreation implies a new state exists.
|
|
if after is None:
|
|
# This might happen if the plan is malformed or in unexpected states, but for safety:
|
|
print(f"WARNING: Resource {address} is being recreated but has no 'after' state to check tags.")
|
|
continue
|
|
|
|
# Tags can be a list or a comma-separated string depending on provider
|
|
tags = after.get('tags', [])
|
|
if tags is None:
|
|
tags = []
|
|
elif isinstance(tags, str):
|
|
tags = [t.strip() for t in tags.split(',')]
|
|
|
|
if 'transient' not in tags:
|
|
print(f"ERROR: Validation Failed! Resource {address} is being recreated but is not marked as 'transient'. Tags: {tags}")
|
|
sys.exit(1)
|
|
|
|
print(f"Validation Successful: Gitea server exists and {runner_count} runners will be present.")
|
|
sys.exit(0)
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("Usage: python validate_plan.py <plan.json>")
|
|
sys.exit(1)
|
|
|
|
validate_plan(sys.argv[1])
|