main.go

package main

import (
	"bytes"
	"context"
	"crypto/sha256"
	"crypto/tls"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"runtime"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	policyManager "github.com/compliance-framework/agent/policy-manager"
	"github.com/compliance-framework/agent/runner"
	"github.com/compliance-framework/agent/runner/proto"
	"github.com/hashicorp/go-hclog"
	goplugin "github.com/hashicorp/go-plugin"
	"github.com/mitchellh/mapstructure"
	"gopkg.in/yaml.v3"
)

const (
	defaultCheckTimeoutSeconds  = 300
	schemaVersionV2             = "v2"
	sourceCloudCustodian        = "cloud-custodian"
	defaultRemotePolicyTimeout  = 30 * time.Second
	defaultMaxRemotePolicyBytes = 1 << 20 // 1 MiB
	evidenceBatchSize           = 100
	nonComplianceMessageField   = "non_compliance_message"
	custodianWatchInterval      = 30 * time.Second
	custodianOutputTailBytes    = 4096
	custodianLogTailMaxSections = 5
)

var lookPath = exec.LookPath
var lookupHost = net.DefaultResolver.LookupHost
var tlsProbeEndpoint = defaultTLSProbeEndpoint

var awsResourceServices = map[string][]string{
	"app-elb":         {"elasticloadbalancing"},
	"backup-plan":     {"backup"},
	"backup-vault":    {"backup"},
	"cache-cluster":   {"elasticache"},
	"distribution":    {"cloudfront"},
	"dynamodb-table":  {"dynamodb"},
	"ebs":             {"ec2"},
	"ec2":             {"ec2"},
	"ecs-service":     {"ecs"},
	"efs":             {"elasticfilesystem"},
	"eks":             {"eks"},
	"firewall":        {"network-firewall"},
	"hostedzone":      {"route53"},
	"iam-group":       {"iam"},
	"iam-policy":      {"iam"},
	"iam-role":        {"iam"},
	"iam-user":        {"iam"},
	"kms-key":         {"kms"},
	"lambda":          {"lambda"},
	"log-group":       {"logs"},
	"rds":             {"rds"},
	"rds-cluster":     {"rds"},
	"s3":              {"s3"},
	"secrets-manager": {"secretsmanager"},
	"sns":             {"sns"},
	"sqs":             {"sqs"},
	"transfer-server": {"transfer"},
	"wafv2":           {"wafv2"},
}

var awsGlobalEndpointServices = map[string]string{
	"cloudfront": "cloudfront.amazonaws.com",
	"iam":        "iam.amazonaws.com",
	"route53":    "route53.amazonaws.com",
}

// PluginConfig receives string-only config from the agent gRPC interface.
type PluginConfig struct {
	PoliciesYAML                        string `mapstructure:"policies_yaml"`
	PoliciesPath                        string `mapstructure:"policies_path"`
	CustodianBinary                     string `mapstructure:"custodian_binary"`
	CustodianDebug                      string `mapstructure:"custodian_debug"`
	CustodianVerbose                    string `mapstructure:"custodian_verbose"`
	CustodianAWSAPITrace                string `mapstructure:"custodian_aws_api_trace"`
	CustodianNetworkDiag                string `mapstructure:"custodian_network_diagnostics"`
	CustodianNetworkDiagnosticEndpoints string `mapstructure:"custodian_network_diagnostic_endpoints"`
	CustodianLogTail                    string `mapstructure:"custodian_log_tail_during_run"`
	AWSRegions                          string `mapstructure:"aws_regions"`
	PolicyLabels                        string `mapstructure:"policy_labels"`
	ResourceIdentityFields              string `mapstructure:"resource_identity_fields"`
	CheckTimeoutSeconds                 string `mapstructure:"check_timeout_seconds"`
	DebugDumpPayloads                   string `mapstructure:"debug_dump_payloads"`
	DebugPayloadOutputDir               string `mapstructure:"debug_payload_output_dir"`
	PreserveArtifacts                   string `mapstructure:"preserve_execution_artifacts"`
}

// ParsedConfig stores normalized and validated values for runtime use.
type ParsedConfig struct {
	PoliciesYAML                        string
	PoliciesPath                        string
	CustodianBinary                     string
	CustodianDebug                      bool
	CustodianVerbose                    bool
	CustodianAWSAPITrace                bool
	CustodianNetworkDiag                bool
	CustodianNetworkDiagnosticEndpoints []string
	CustodianLogTail                    bool
	AWSRegions                          []string
	PolicyLabels                        map[string]string
	ResourceIdentityFields              map[string][]string
	CheckTimeout                        time.Duration
	DebugDumpPayloads                   bool
	DebugPayloadOutputDir               string
	PreserveArtifacts                   bool
}

func (c *PluginConfig) Parse() (*ParsedConfig, error) {
	inlineYAML := strings.TrimSpace(c.PoliciesYAML)
	policiesPath := strings.TrimSpace(c.PoliciesPath)

	if inlineYAML == "" && policiesPath == "" {
		return nil, errors.New("either policies_yaml or policies_path is required")
	}

	policyLabels := map[string]string{}
	if strings.TrimSpace(c.PolicyLabels) != "" {
		if err := json.Unmarshal([]byte(c.PolicyLabels), &policyLabels); err != nil {
			return nil, fmt.Errorf("could not parse policy_labels: %w", err)
		}
	}

	resourceIdentityFields := map[string][]string{}
	if strings.TrimSpace(c.ResourceIdentityFields) != "" {
		if err := json.Unmarshal([]byte(c.ResourceIdentityFields), &resourceIdentityFields); err != nil {
			return nil, fmt.Errorf("could not parse resource_identity_fields: %w", err)
		}
		normalizedIdentityFields := make(map[string][]string, len(resourceIdentityFields))
		for resourceType, fields := range resourceIdentityFields {
			trimmedType := strings.TrimSpace(resourceType)
			if trimmedType == "" {
				return nil, errors.New("resource_identity_fields cannot contain an empty resource type")
			}
			normalizedFields := make([]string, 0, len(fields))
			for _, field := range fields {
				field = strings.TrimSpace(field)
				if field != "" {
					normalizedFields = append(normalizedFields, field)
				}
			}
			if len(normalizedFields) == 0 {
				return nil, fmt.Errorf("resource_identity_fields for %q must include at least one field", trimmedType)
			}
			if _, exists := normalizedIdentityFields[trimmedType]; exists {
				return nil, fmt.Errorf("resource_identity_fields contains duplicate resource type %q after trimming whitespace", trimmedType)
			}
			normalizedIdentityFields[trimmedType] = normalizedFields
		}
		resourceIdentityFields = normalizedIdentityFields
	}

	checkTimeoutSeconds := defaultCheckTimeoutSeconds
	if strings.TrimSpace(c.CheckTimeoutSeconds) != "" {
		parsedTimeout, err := strconv.Atoi(c.CheckTimeoutSeconds)
		if err != nil {
			return nil, fmt.Errorf("check_timeout_seconds must be a positive integer: %w", err)
		}
		if parsedTimeout <= 0 {
			return nil, errors.New("check_timeout_seconds must be greater than 0")
		}
		checkTimeoutSeconds = parsedTimeout
	}

	binary := strings.TrimSpace(c.CustodianBinary)
	if binary == "" {
		binary = "custodian"
	}

	resolvedBinary, err := lookPath(binary)
	if err != nil {
		return nil, fmt.Errorf("could not resolve custodian binary %q: %w", binary, err)
	}

	awsRegions := parseDelimitedList(c.AWSRegions)
	networkDiagnosticEndpoints := parseDelimitedList(c.CustodianNetworkDiagnosticEndpoints)

	custodianDebug, err := parseOptionalBool("custodian_debug", c.CustodianDebug)
	if err != nil {
		return nil, err
	}

	custodianVerbose, err := parseOptionalBool("custodian_verbose", c.CustodianVerbose)
	if err != nil {
		return nil, err
	}

	custodianAWSAPITrace, err := parseOptionalBool("custodian_aws_api_trace", c.CustodianAWSAPITrace)
	if err != nil {
		return nil, err
	}

	custodianNetworkDiag, err := parseOptionalBool("custodian_network_diagnostics", c.CustodianNetworkDiag)
	if err != nil {
		return nil, err
	}

	custodianLogTail, err := parseOptionalBool("custodian_log_tail_during_run", c.CustodianLogTail)
	if err != nil {
		return nil, err
	}

	debugDumpPayloads, err := parseOptionalBool("debug_dump_payloads", c.DebugDumpPayloads)
	if err != nil {
		return nil, err
	}

	preserveArtifacts, err := parseOptionalBool("preserve_execution_artifacts", c.PreserveArtifacts)
	if err != nil {
		return nil, err
	}

	debugPayloadOutputDir := strings.TrimSpace(c.DebugPayloadOutputDir)
	if debugPayloadOutputDir != "" {
		debugDumpPayloads = true
	}
	if debugDumpPayloads && debugPayloadOutputDir == "" {
		debugPayloadOutputDir = "debug-standardized-payloads"
	}

	return &ParsedConfig{
		PoliciesYAML:                        inlineYAML,
		PoliciesPath:                        policiesPath,
		CustodianBinary:                     resolvedBinary,
		CustodianDebug:                      custodianDebug,
		CustodianVerbose:                    custodianVerbose,
		CustodianAWSAPITrace:                custodianAWSAPITrace,
		CustodianNetworkDiag:                custodianNetworkDiag,
		CustodianNetworkDiagnosticEndpoints: networkDiagnosticEndpoints,
		CustodianLogTail:                    custodianLogTail,
		AWSRegions:                          awsRegions,
		PolicyLabels:                        policyLabels,
		ResourceIdentityFields:              resourceIdentityFields,
		CheckTimeout:                        time.Duration(checkTimeoutSeconds) * time.Second,
		DebugDumpPayloads:                   debugDumpPayloads,
		DebugPayloadOutputDir:               debugPayloadOutputDir,
		PreserveArtifacts:                   preserveArtifacts,
	}, nil
}

func parseOptionalBool(name, value string) (bool, error) {
	value = strings.TrimSpace(value)
	if value == "" {
		return false, nil
	}
	parsed, err := strconv.ParseBool(value)
	if err != nil {
		return false, fmt.Errorf("%s must be a boolean value: %w", name, err)
	}
	return parsed, nil
}

func parseDelimitedList(value string) []string {
	parts := strings.FieldsFunc(value, func(r rune) bool {
		return r == ',' || r == ' ' || r == '\n' || r == '\t' || r == '\r'
	})
	return compactUniqueStrings(parts)
}

// CustodianCheck represents a single Cloud Custodian policy entry used as one check iteration.
type CustodianCheck struct {
	Index       int
	Name        string
	Resource    string
	Provider    string
	RawPolicy   map[string]interface{}
	ParseErrors []string
}

// CustodianExecutionRequest contains execution-time settings for one check run.
type CustodianExecutionRequest struct {
	BinaryPath                 string
	Check                      CustodianCheck
	Timeout                    time.Duration
	OutputDir                  string
	Debug                      bool
	Verbose                    bool
	AWSRegions                 []string
	AWSAPITrace                bool
	NetworkDiagnostics         bool
	NetworkDiagnosticEndpoints []string
	LogTailDuringRun           bool
}

// CustodianExecutionResult captures runtime output and artifacts from one check run.
type CustodianExecutionResult struct {
	StartedAt          time.Time
	EndedAt            time.Time
	ExitCode           int
	Stdout             string
	Stderr             string
	Error              string
	Errors             []string
	Err                error
	Resources          []interface{}
	ArtifactPath       string
	ResourcesPath      string
	LogPaths           []string
	DiagnosticWarnings []string
}

// CustodianExecutor runs one Cloud Custodian check and captures execution artifacts.
type CustodianExecutor interface {
	Execute(ctx context.Context, req CustodianExecutionRequest) CustodianExecutionResult
}

// CommandCustodianExecutor executes the custodian CLI.
type CommandCustodianExecutor struct {
	Logger hclog.Logger
}

type lockedBuffer struct {
	mu  sync.Mutex
	buf bytes.Buffer
}

func (b *lockedBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.Write(p)
}

func (b *lockedBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.String()
}

func (b *lockedBuffer) Len() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.buf.Len()
}

func (b *lockedBuffer) Tail(maxBytes int) string {
	b.mu.Lock()
	defer b.mu.Unlock()

	content := b.buf.Bytes()
	if maxBytes <= 0 || len(content) <= maxBytes {
		return string(content)
	}
	return string(content[len(content)-maxBytes:])
}

func custodianDiagnosticInterval(timeout time.Duration) time.Duration {
	if timeout <= 0 {
		return custodianWatchInterval
	}
	interval := custodianWatchInterval
	if timeout < 2*interval {
		interval = timeout / 2
	}
	if interval < time.Second {
		return time.Second
	}
	return interval
}

func (e *CommandCustodianExecutor) Execute(ctx context.Context, req CustodianExecutionRequest) CustodianExecutionResult {
	e.Logger.Debug("Starting cloud custodian execution",
		"check_name", req.Check.Name,
		"check_index", req.Check.Index,
		"resource", req.Check.Resource,
		"provider", req.Check.Provider,
		"binary", req.BinaryPath,
		"timeout", req.Timeout.String(),
		"output_dir", req.OutputDir,
		"aws_regions", req.AWSRegions,
	)
	result := CustodianExecutionResult{
		StartedAt:    time.Now().UTC(),
		ExitCode:     -1,
		Resources:    []interface{}{},
		Errors:       []string{},
		ArtifactPath: req.OutputDir,
	}

	if err := os.MkdirAll(req.OutputDir, 0o755); err != nil {
		result.Err = fmt.Errorf("failed to create output directory: %w", err)
		result.Error = result.Err.Error()
		result.Errors = []string{result.Error}
		e.Logger.Error("Failed creating output directory for check", "check_name", req.Check.Name, "error", result.Error)
		result.EndedAt = time.Now().UTC()
		return result
	}
	e.Logger.Trace("Created output directory for check", "check_name", req.Check.Name, "output_dir", req.OutputDir)

	policyDocument := map[string]interface{}{
		"policies": []map[string]interface{}{custodianPolicyForExecution(req.Check.RawPolicy)},
	}
	policyContent, err := yaml.Marshal(policyDocument)
	if err != nil {
		result.Err = fmt.Errorf("failed to marshal single policy document: %w", err)
		result.Error = result.Err.Error()
		result.Errors = []string{result.Error}
		e.Logger.Error("Failed marshaling single policy yaml for check", "check_name", req.Check.Name, "error", result.Error)
		result.EndedAt = time.Now().UTC()
		return result
	}

	policyPath := filepath.Join(req.OutputDir, "policy.yaml")
	if err := os.WriteFile(policyPath, policyContent, 0o600); err != nil {
		result.Err = fmt.Errorf("failed to write single policy file: %w", err)
		result.Error = result.Err.Error()
		result.Errors = []string{result.Error}
		e.Logger.Error("Failed writing single policy file for check", "check_name", req.Check.Name, "policy_path", policyPath, "error", result.Error)
		result.EndedAt = time.Now().UTC()
		return result
	}
	e.Logger.Trace("Wrote single policy file", "check_name", req.Check.Name, "policy_path", policyPath)

	runCtx, cancel := context.WithTimeout(ctx, req.Timeout)
	defer cancel()

	regions := req.AWSRegions
	if strings.EqualFold(req.Check.Provider, "aws") {
		// Ensure AWS policies evaluate across all regions by default while
		// allowing operators to narrow problematic service/region scans.
		if len(regions) == 0 {
			regions = []string{"all"}
		}
	}
	if req.NetworkDiagnostics && strings.EqualFold(req.Check.Provider, "aws") {
		diagnostics, diagErr := e.runAWSEndpointDiagnostics(runCtx, req)
		result.DiagnosticWarnings = append(result.DiagnosticWarnings, diagnostics.executionWarnings(req.Check)...)
		if diagErr != nil {
			result.Err = fmt.Errorf("aws endpoint network diagnostics failed: %w", diagErr)
			result.Errors = []string{result.Err.Error()}
			result.Error = executionErrorString(result)
			result.EndedAt = time.Now().UTC()
			e.Logger.Error("Skipping custodian command because AWS endpoint diagnostics failed", "check_name", req.Check.Name, "error", result.Error)
			return result
		}
		if availableRegions, ok := diagnostics.availableAWSRegions(regions); ok {
			if len(availableRegions) == 0 {
				result.Err = fmt.Errorf("cloud custodian policy %s could not be checked because no AWS service endpoints were reachable for resource %s in requested regions %s", req.Check.Name, req.Check.Resource, strings.Join(regions, ","))
				result.Errors = append([]string{result.Err.Error()}, result.Errors...)
				result.Error = executionErrorString(result)
				e.Logger.Warn("Skipping custodian command because no AWS service endpoints were reachable for policy",
					"check_name", req.Check.Name,
					"resource", req.Check.Resource,
					"aws_regions", regions,
					"unavailable_endpoint_count", len(diagnostics.Failures),
				)
				result.EndedAt = time.Now().UTC()
				return result
			}
			if !slices.Equal(regions, availableRegions) {
				e.Logger.Warn("Running custodian command only for AWS regions with reachable service endpoints",
					"check_name", req.Check.Name,
					"resource", req.Check.Resource,
					"requested_aws_regions", regions,
					"available_aws_regions", availableRegions,
					"unavailable_endpoint_count", len(diagnostics.Failures),
				)
				regions = availableRegions
			}
		}
	}

	args := []string{"run"}
	if req.Debug {
		args = append(args, "--debug")
	}
	if req.Verbose {
		args = append(args, "-v")
	}
	args = append(args, "--dryrun", "-s", req.OutputDir, policyPath)
	if strings.EqualFold(req.Check.Provider, "aws") {
		for _, region := range regions {
			args = append(args, "--region", region)
		}
	}
	cmd := exec.CommandContext(runCtx, req.BinaryPath, args...)
	cmd.Env = os.Environ()
	if req.AWSAPITrace {
		traceDir, traceLogPath, traceErr := setupCustodianAWSAPITrace(req.OutputDir)
		if traceErr != nil {
			e.Logger.Warn("Failed setting up custodian AWS API trace", "check_name", req.Check.Name, "error", traceErr)
		} else {
			cmd.Env = upsertEnv(cmd.Env, "PYTHONPATH", prependPathList(traceDir, os.Getenv("PYTHONPATH")))
			cmd.Env = upsertEnv(cmd.Env, "CCF_CUSTODIAN_AWS_API_TRACE_LOG", traceLogPath)
			cmd.Env = upsertEnv(cmd.Env, "PYTHONUNBUFFERED", "1")
			e.Logger.Info("Custodian AWS API trace enabled", "check_name", req.Check.Name, "trace_log_path", traceLogPath, "pythonpath_dir", traceDir)
		}
	}
	e.Logger.Debug("Executing custodian command",
		"check_name", req.Check.Name,
		"command", req.BinaryPath,
		"args", args,
	)
	stdoutBuf := &lockedBuffer{}
	stderrBuf := &lockedBuffer{}
	cmd.Stdout = stdoutBuf
	cmd.Stderr = stderrBuf

	err = cmd.Start()
	if err == nil {
		pid := -1
		if cmd.Process != nil {
			pid = cmd.Process.Pid
		}
		e.Logger.Info("Custodian command process started",
			"check_name", req.Check.Name,
			"pid", pid,
		)

		waitDone := make(chan error, 1)
		e.Logger.Debug("Setting up custodian command wait channel",
			"check_name", req.Check.Name,
			"pid", pid,
			"buffered", true,
			"capacity", cap(waitDone),
		)
		go func() {
			waitDone <- cmd.Wait()
		}()

		diagnosticInterval := custodianDiagnosticInterval(req.Timeout)
		ticker := time.NewTicker(diagnosticInterval)
		defer ticker.Stop()
		contextDoneLogged := false
		runCtxDone := runCtx.Done()
		lastCustodianLogTail := ""
		logTailCache := &custodianLogTailCache{}
		e.Logger.Debug("Starting custodian command monitor loop",
			"check_name", req.Check.Name,
			"pid", pid,
			"diagnostic_interval", diagnosticInterval.String(),
			"timeout", req.Timeout.String(),
		)

		for {
			select {
			case err = <-waitDone:
				e.Logger.Info("Custodian command wait completed",
					"check_name", req.Check.Name,
					"pid", pid,
					"elapsed", time.Since(result.StartedAt).Round(time.Second).String(),
					"wait_error", err,
					"context_error", runCtx.Err(),
					"stdout_len", stdoutBuf.Len(),
					"stderr_len", stderrBuf.Len(),
				)
				goto commandFinished
			case <-ticker.C:
				elapsed := time.Since(result.StartedAt).Round(time.Second).String()
				remaining := ""
				if deadline, ok := runCtx.Deadline(); ok {
					remaining = time.Until(deadline).Round(time.Second).String()
				}
				e.Logger.Info("Custodian command still running",
					"check_name", req.Check.Name,
					"pid", pid,
					"elapsed", elapsed,
					"remaining", remaining,
					"timeout", req.Timeout.String(),
					"stdout_len", stdoutBuf.Len(),
					"stderr_len", stderrBuf.Len(),
				)
				if req.NetworkDiagnostics {
					e.logCustodianProcessSockets(pid, req.Check.Name)
				}
				if req.LogTailDuringRun {
					lastCustodianLogTail = e.logCustodianRunLogTail(req.OutputDir, req.Check.Name, lastCustodianLogTail, logTailCache)
				}
			case <-runCtxDone:
				if !contextDoneLogged {
					e.Logger.Warn("Custodian command context done while process is still running",
						"check_name", req.Check.Name,
						"pid", pid,
						"elapsed", time.Since(result.StartedAt).Round(time.Second).String(),
						"timeout", req.Timeout.String(),
						"context_error", runCtx.Err(),
						"stdout_len", stdoutBuf.Len(),
						"stderr_len", stderrBuf.Len(),
						"stderr_tail", stderrBuf.Tail(custodianOutputTailBytes),
					)
					if req.NetworkDiagnostics {
						e.logCustodianProcessSockets(pid, req.Check.Name)
					}
					if req.LogTailDuringRun {
						lastCustodianLogTail = e.logCustodianRunLogTail(req.OutputDir, req.Check.Name, lastCustodianLogTail, logTailCache)
					}
					contextDoneLogged = true
				}
				runCtxDone = nil
			}
		}
	} else {
		e.Logger.Warn("Failed to start custodian command process",
			"check_name", req.Check.Name,
			"command", req.BinaryPath,
			"args", args,
			"error", err,
		)
	}

commandFinished:
	result.Stdout = stdoutBuf.String()
	result.Stderr = stderrBuf.String()
	if cmd.ProcessState != nil {
		result.ExitCode = cmd.ProcessState.ExitCode()
	}
	e.Logger.Debug("Custodian command finished",
		"check_name", req.Check.Name,
		"exit_code", result.ExitCode,
		"stdout_len", len(result.Stdout),
		"stderr_len", len(result.Stderr),
	)

	resourcesPath, resources, resourcesErr := readResourcesArtifact(req.OutputDir)
	result.ResourcesPath = resourcesPath
	if resources != nil {
		result.Resources = resources
	}
	var logTail string
	var logErr error
	if req.LogTailDuringRun || err != nil || runCtx.Err() != nil || resourcesErr != nil {
		logPaths, tail, readErr := readCustodianLogArtifacts(req.OutputDir, custodianOutputTailBytes)
		result.LogPaths = logPaths
		logTail = tail
		logErr = readErr
	}

	if err != nil {
		result.Err = fmt.Errorf("custodian execution failed: %w", err)
		result.Errors = append(result.Errors, result.Err.Error())
		if result.Stderr != "" {
			result.Errors = append(result.Errors, result.Stderr)
		}
	}
	if runErr := runCtx.Err(); runErr != nil {
		// Avoid duplicating context timeout/cancel errors when cmd.Run already
		// returned an error that wraps the same context failure.
		if err == nil || !errors.Is(err, runErr) {
			result.Err = errors.Join(result.Err, runErr)
			result.Errors = append(result.Errors, runErr.Error())
		}
	}
	if resourcesErr != nil {
		result.Err = errors.Join(result.Err, resourcesErr)
		result.Errors = append(result.Errors, resourcesErr.Error())
	}
	if logErr != nil {
		if result.Err != nil {
			result.Err = errors.Join(result.Err, logErr)
			result.Errors = append(result.Errors, logErr.Error())
		} else {
			e.Logger.Warn("Failed collecting custodian log artifacts after successful execution",
				"check_name", req.Check.Name,
				"error", logErr,
			)
		}
	}
	if result.Err != nil && logTail != "" {
		result.Errors = append(result.Errors, logTail)
	}

	if result.Err != nil {
		result.Error = executionErrorString(result)
		e.Logger.Warn("Custodian execution completed with errors",
			"check_name", req.Check.Name,
			"error_count", len(result.Errors),
			"errors", result.Errors,
		)
	} else {
		e.Logger.Debug("Custodian execution completed successfully",
			"check_name", req.Check.Name,
			"resource_count", len(result.Resources),
			"resources_path", result.ResourcesPath,
		)
	}

	result.EndedAt = time.Now().UTC()
	return result
}

func setupCustodianAWSAPITrace(outputDir string) (string, string, error) {
	traceDir := filepath.Join(outputDir, "ccf-custodian-python-trace")
	if err := os.MkdirAll(traceDir, 0o700); err != nil {
		return "", "", err
	}
	traceLogPath := filepath.Join(outputDir, "custodian-aws-api-trace.jsonl")
	traceLog, err := os.OpenFile(traceLogPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		return "", "", err
	}
	if err := traceLog.Close(); err != nil {
		return "", "", err
	}
	if err := os.Chmod(traceLogPath, 0o600); err != nil {
		return "", "", err
	}
	siteCustomizePath := filepath.Join(traceDir, "sitecustomize.py")
	siteCustomize := `import json
import os
import sys
import time

def _ccf_trace_write(record):
    record["timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    line = json.dumps(record, sort_keys=True, default=str)
    path = os.environ.get("CCF_CUSTODIAN_AWS_API_TRACE_LOG")
    if path:
        try:
            with open(path, "a", encoding="utf-8") as handle:
                handle.write(line + "\n")
        except Exception as exc:
            sys.stderr.write("[CCF_AWS_API_TRACE] trace_write_failed " + type(exc).__name__ + ": " + str(exc) + "\n")
    sys.stderr.write("[CCF_AWS_API_TRACE] " + line + "\n")
    sys.stderr.flush()

try:
    from botocore.client import BaseClient
    _ccf_original_make_api_call = BaseClient._make_api_call

    def _ccf_make_api_call(self, operation_name, api_params):
        meta = getattr(self, "meta", None)
        service = ""
        endpoint = ""
        region = ""
        if meta is not None:
            service_model = getattr(meta, "service_model", None)
            service = getattr(service_model, "service_name", "")
            endpoint = getattr(meta, "endpoint_url", "")
            region = getattr(meta, "region_name", "")
        start = time.time()
        _ccf_trace_write({
            "event": "aws_api_start",
            "service": service,
            "operation": operation_name,
            "endpoint": endpoint,
            "region": region,
        })
        try:
            response = _ccf_original_make_api_call(self, operation_name, api_params)
        except Exception as exc:
            _ccf_trace_write({
                "event": "aws_api_error",
                "service": service,
                "operation": operation_name,
                "endpoint": endpoint,
                "region": region,
                "elapsed_ms": int((time.time() - start) * 1000),
                "error_type": type(exc).__name__,
                "error": str(exc),
            })
            raise
        _ccf_trace_write({
            "event": "aws_api_end",
            "service": service,
            "operation": operation_name,
            "endpoint": endpoint,
            "region": region,
            "elapsed_ms": int((time.time() - start) * 1000),
        })
        return response

    BaseClient._make_api_call = _ccf_make_api_call
    _ccf_trace_write({"event": "aws_api_trace_installed"})
except Exception as exc:
    sys.stderr.write("[CCF_AWS_API_TRACE] install_failed " + type(exc).__name__ + ": " + str(exc) + "\n")
    sys.stderr.flush()
`
	if err := os.WriteFile(siteCustomizePath, []byte(siteCustomize), 0o600); err != nil {
		return "", "", err
	}
	return traceDir, traceLogPath, nil
}

func upsertEnv(env []string, key, value string) []string {
	prefix := key + "="
	for index, entry := range env {
		if strings.HasPrefix(entry, prefix) {
			env[index] = prefix + value
			return env
		}
	}
	return append(env, prefix+value)
}

func prependPathList(path, existing string) string {
	if strings.TrimSpace(existing) == "" {
		return path
	}
	return path + string(os.PathListSeparator) + existing
}

type networkDiagnosticEndpoint struct {
	Host       string
	Port       string
	ServerName string
	Source     string
	Service    string
	Region     string
}

type awsEndpointDiagnosticFailure struct {
	Endpoint networkDiagnosticEndpoint
	Stage    string
	Err      error
}

type awsEndpointDiagnosticResult struct {
	Failures             []awsEndpointDiagnosticFailure
	regionProbeSucceeded map[string]bool
	regionProbeFailed    map[string]bool
}

func (r awsEndpointDiagnosticResult) availableAWSRegions(regions []string) ([]string, bool) {
	if len(r.regionProbeSucceeded) == 0 && len(r.regionProbeFailed) == 0 {
		return nil, false
	}
	available := make([]string, 0, len(regions))
	for _, region := range regions {
		region = strings.TrimSpace(region)
		if region == "" || strings.EqualFold(region, "all") {
			continue
		}
		if r.regionProbeSucceeded[region] && !r.regionProbeFailed[region] {
			available = append(available, region)
		}
	}
	return available, true
}

func (e *CommandCustodianExecutor) runAWSEndpointDiagnostics(ctx context.Context, req CustodianExecutionRequest) (awsEndpointDiagnosticResult, error) {
	result := awsEndpointDiagnosticResult{
		Failures:             []awsEndpointDiagnosticFailure{},
		regionProbeSucceeded: map[string]bool{},
		regionProbeFailed:    map[string]bool{},
	}
	endpoints, knownResource, endpointErr := awsDiagnosticEndpointsForCheck(req.Check.Resource, req.AWSRegions, req.NetworkDiagnosticEndpoints)
	if endpointErr != nil {
		e.Logger.Error("AWS endpoint diagnostics configuration is invalid", "check_name", req.Check.Name, "resource", req.Check.Resource, "error", endpointErr)
		return result, endpointErr
	}
	if !knownResource && len(endpoints) == 0 {
		e.Logger.Warn("Skipping AWS endpoint diagnostics because resource service is not mapped and no explicit endpoints are configured", "check_name", req.Check.Name, "resource", req.Check.Resource)
		return result, nil
	}
	if len(endpoints) == 0 {
		e.Logger.Warn("Skipping AWS endpoint diagnostics because no concrete endpoint hosts are available; configure aws_regions or custodian_network_diagnostic_endpoints for preflight probes", "check_name", req.Check.Name, "resource", req.Check.Resource, "aws_regions", req.AWSRegions)
		return result, nil
	}
	if !knownResource {
		e.Logger.Warn("AWS endpoint diagnostics will use only configured endpoints because resource service is not mapped", "check_name", req.Check.Name, "resource", req.Check.Resource)
	}

	for _, endpoint := range endpoints {
		if err := ctx.Err(); err != nil {
			return result, err
		}
		lookupCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
		lookupStarted := time.Now()
		ips, err := lookupHost(lookupCtx, endpoint.Host)
		cancel()
		if err != nil {
			result.recordFailure(endpoint, "DNS lookup", err)
			e.Logger.Warn("AWS endpoint diagnostics detected an unreachable service endpoint; evaluation may be partial",
				"check_name", req.Check.Name,
				"resource", req.Check.Resource,
				"service", endpoint.Service,
				"region", endpoint.Region,
				"host", endpoint.Host,
				"stage", "DNS lookup",
				"source", endpoint.Source,
				"elapsed", time.Since(lookupStarted).Round(time.Millisecond).String(),
				"error", err,
			)
			continue
		} else {
			e.Logger.Info("AWS endpoint DNS lookup succeeded", "check_name", req.Check.Name, "host", endpoint.Host, "source", endpoint.Source, "ips", ips, "elapsed", time.Since(lookupStarted).Round(time.Millisecond).String())
		}

		tlsStarted := time.Now()
		tlsResult, err := tlsProbeEndpoint(ctx, endpoint)
		if err != nil {
			result.recordFailure(endpoint, "TLS handshake", err)
			e.Logger.Warn("AWS endpoint diagnostics detected an unreachable service endpoint; evaluation may be partial",
				"check_name", req.Check.Name,
				"resource", req.Check.Resource,
				"service", endpoint.Service,
				"region", endpoint.Region,
				"host", endpoint.Host,
				"stage", "TLS handshake",
				"port", endpoint.Port,
				"server_name", endpoint.ServerName,
				"source", endpoint.Source,
				"elapsed", time.Since(tlsStarted).Round(time.Millisecond).String(),
				"error", err,
			)
			continue
		}
		if endpoint.Source == "aws-service" && endpoint.Region != "" {
			if result.regionProbeSucceeded == nil {
				result.regionProbeSucceeded = map[string]bool{}
			}
			result.regionProbeSucceeded[endpoint.Region] = true
		}
		e.Logger.Info("AWS endpoint TLS probe succeeded", "check_name", req.Check.Name, "host", endpoint.Host, "port", endpoint.Port, "server_name", endpoint.ServerName, "source", endpoint.Source, "remote_addr", tlsResult.RemoteAddr, "tls_version", tlsResult.TLSVersion, "elapsed", time.Since(tlsStarted).Round(time.Millisecond).String())
	}
	return result, nil
}

func (r *awsEndpointDiagnosticResult) recordFailure(endpoint networkDiagnosticEndpoint, stage string, err error) {
	if r.regionProbeFailed == nil {
		r.regionProbeFailed = map[string]bool{}
	}
	r.Failures = append(r.Failures, awsEndpointDiagnosticFailure{
		Endpoint: endpoint,
		Stage:    stage,
		Err:      err,
	})
	if endpoint.Source == "aws-service" && endpoint.Region != "" {
		r.regionProbeFailed[endpoint.Region] = true
	}
}

func (r awsEndpointDiagnosticResult) executionWarnings(check CustodianCheck) []string {
	if len(r.Failures) == 0 {
		return nil
	}
	messages := make([]string, 0, len(r.Failures))
	for _, failure := range r.Failures {
		endpoint := failure.Endpoint
		service := endpoint.Service
		if service == "" {
			service = endpoint.Source
		}
		region := endpoint.Region
		if region == "" {
			region = "global"
		}
		messages = append(messages, fmt.Sprintf(
			"unreachable AWS service endpoint %s.%s (%s:%s) detected while evaluating cloud custodian policy %s; evaluation may be partial: %s failed: %v",
			service,
			region,
			endpoint.Host,
			endpoint.Port,
			check.Name,
			failure.Stage,
			failure.Err,
		))
	}
	return messages
}

func executionErrorString(result CustodianExecutionResult) string {
	messages := make([]string, 0, len(result.Errors)+len(result.DiagnosticWarnings))
	seen := map[string]bool{}
	for _, values := range [][]string{result.Errors, result.DiagnosticWarnings} {
		for _, message := range values {
			message = strings.TrimSpace(message)
			if message == "" || seen[message] {
				continue
			}
			messages = append(messages, message)
			seen[message] = true
		}
	}
	return strings.Join(messages, "; ")
}

func awsEndpointHostsForCheck(resource string, regions []string) ([]string, bool) {
	endpoints, knownResource, err := awsDiagnosticEndpointsForCheck(resource, regions, nil)
	if err != nil {
		return nil, knownResource
	}
	hosts := make([]string, 0, len(endpoints))
	for _, endpoint := range endpoints {
		hosts = append(hosts, endpoint.Host)
	}
	return hosts, knownResource
}

func awsDiagnosticEndpointsForCheck(resource string, regions []string, configuredEndpoints []string) ([]networkDiagnosticEndpoint, bool, error) {
	services, knownResource := awsServicesForResource(resource)
	endpoints := make([]networkDiagnosticEndpoint, 0)
	for _, endpointValue := range configuredEndpoints {
		endpoint, err := parseNetworkDiagnosticEndpoint(endpointValue)
		if err != nil {
			return nil, knownResource, err
		}
		endpoints = append(endpoints, endpoint)
	}
	if !knownResource {
		return compactUniqueNetworkDiagnosticEndpoints(endpoints), false, nil
	}

	diagnosticRegions := make([]string, 0, len(regions))
	for _, region := range regions {
		region = strings.TrimSpace(region)
		if region == "" || strings.EqualFold(region, "all") {
			continue
		}
		diagnosticRegions = append(diagnosticRegions, region)
	}
	if len(diagnosticRegions) == 0 {
		return compactUniqueNetworkDiagnosticEndpoints(endpoints), true, nil
	}

	for _, region := range diagnosticRegions {
		for _, service := range services {
			host := awsServiceEndpointHost(service, region)
			endpointRegion := region
			if _, ok := awsGlobalEndpointServices[service]; ok {
				endpointRegion = ""
			}
			endpoints = append(endpoints, networkDiagnosticEndpoint{
				Host:       host,
				Port:       "443",
				ServerName: host,
				Source:     "aws-service",
				Service:    service,
				Region:     endpointRegion,
			})
		}
	}
	return compactUniqueNetworkDiagnosticEndpoints(endpoints), true, nil
}

func awsServiceEndpointHost(service string, region string) string {
	service = strings.TrimSpace(service)
	region = strings.TrimSpace(region)
	partition := awsPartitionForRegion(region)
	dnsSuffix := awsEndpointDNSSuffixForPartition(partition)
	if _, ok := awsGlobalEndpointServices[service]; ok && partition == "aws-cn" {
		return fmt.Sprintf("%s.%s", service, dnsSuffix)
	}
	if _, ok := awsGlobalEndpointServices[service]; ok && partition == "aws-us-gov" {
		return fmt.Sprintf("%s.us-gov.%s", service, dnsSuffix)
	}
	if host, ok := awsGlobalEndpointServices[service]; ok {
		return host
	}
	return fmt.Sprintf("%s.%s.%s", service, region, dnsSuffix)
}

func awsEndpointDNSSuffixForPartition(partition string) string {
	if partition == "aws-cn" {
		return "amazonaws.com.cn"
	}
	return "amazonaws.com"
}

func parseNetworkDiagnosticEndpoint(value string) (networkDiagnosticEndpoint, error) {
	original := strings.TrimSpace(value)
	if original == "" {
		return networkDiagnosticEndpoint{}, errors.New("network diagnostic endpoint cannot be empty")
	}
	parseValue := original
	if !strings.Contains(parseValue, "://") {
		parseValue = "https://" + parseValue
	}
	parsed, err := url.Parse(parseValue)
	if err != nil {
		return networkDiagnosticEndpoint{}, fmt.Errorf("failed to parse network diagnostic endpoint %q: %w", original, err)
	}
	if parsed.Scheme != "" && !strings.EqualFold(parsed.Scheme, "https") {
		return networkDiagnosticEndpoint{}, fmt.Errorf("network diagnostic endpoint %q has unsupported scheme %q: only https endpoints are supported", original, parsed.Scheme)
	}
	host := strings.TrimSpace(parsed.Hostname())
	if host == "" {
		return networkDiagnosticEndpoint{}, fmt.Errorf("network diagnostic endpoint %q does not include a host", original)
	}
	port := parsed.Port()
	if port == "" {
		port = "443"
	}
	portNumber, err := strconv.Atoi(port)
	if err != nil {
		return networkDiagnosticEndpoint{}, fmt.Errorf("network diagnostic endpoint %q has invalid port %q: %w", original, port, err)
	}
	if portNumber < 1 || portNumber > 65535 {
		return networkDiagnosticEndpoint{}, fmt.Errorf("network diagnostic endpoint %q has invalid port %q: must be between 1 and 65535", original, port)
	}
	return networkDiagnosticEndpoint{
		Host:       strings.ToLower(host),
		Port:       port,
		ServerName: strings.ToLower(host),
		Source:     networkDiagnosticEndpointSource(host),
	}, nil
}

func networkDiagnosticEndpointSource(host string) string {
	host = strings.ToLower(strings.TrimSpace(host))
	if strings.HasSuffix(host, ".vpce.amazonaws.com") || strings.HasSuffix(host, ".vpce.amazonaws.com.cn") {
		return "aws-vpc-endpoint"
	}
	return "configured"
}

func compactUniqueNetworkDiagnosticEndpoints(values []networkDiagnosticEndpoint) []networkDiagnosticEndpoint {
	seen := map[string]int{}
	result := make([]networkDiagnosticEndpoint, 0, len(values))
	for _, value := range values {
		key := strings.ToLower(net.JoinHostPort(value.Host, value.Port))
		if value.Host == "" || value.Port == "" {
			continue
		}
		if existingIndex, ok := seen[key]; ok {
			if result[existingIndex].Source != "aws-service" && value.Source == "aws-service" {
				result[existingIndex] = value
			}
			continue
		}
		seen[key] = len(result)
		result = append(result, value)
	}
	return result
}

type tlsProbeResult struct {
	RemoteAddr string
	TLSVersion string
}

func defaultTLSProbeEndpoint(ctx context.Context, endpoint networkDiagnosticEndpoint) (tlsProbeResult, error) {
	if err := ctx.Err(); err != nil {
		return tlsProbeResult{}, err
	}
	netDialer := &net.Dialer{Timeout: 5 * time.Second}
	if deadline, ok := ctx.Deadline(); ok {
		if remaining := time.Until(deadline); remaining > 0 && remaining < netDialer.Timeout {
			netDialer.Timeout = remaining
		}
	}
	tlsDialer := &tls.Dialer{
		NetDialer: netDialer,
		Config:    &tls.Config{ServerName: endpoint.ServerName, MinVersion: tls.VersionTLS12},
	}
	conn, err := tlsDialer.DialContext(ctx, "tcp", net.JoinHostPort(endpoint.Host, endpoint.Port))
	if err != nil {
		return tlsProbeResult{}, err
	}
	defer conn.Close()

	tlsConn, ok := conn.(*tls.Conn)
	if !ok {
		return tlsProbeResult{}, errors.New("TLS dial returned non-TLS connection")
	}
	state := tlsConn.ConnectionState()
	return tlsProbeResult{
		RemoteAddr: conn.RemoteAddr().String(),
		TLSVersion: tlsVersionString(state.Version),
	}, nil
}

func awsServicesForResource(resource string) ([]string, bool) {
	resource = strings.TrimPrefix(strings.TrimSpace(resource), "aws.")
	services, ok := awsResourceServices[resource]
	if !ok {
		return nil, false
	}
	return compactUniqueStrings(services), true
}

func tlsVersionString(version uint16) string {
	switch version {
	case tls.VersionTLS10:
		return "TLS1.0"
	case tls.VersionTLS11:
		return "TLS1.1"
	case tls.VersionTLS12:
		return "TLS1.2"
	case tls.VersionTLS13:
		return "TLS1.3"
	default:
		return fmt.Sprintf("0x%x", version)
	}
}

func (e *CommandCustodianExecutor) logCustodianProcessSockets(pid int, checkName string) {
	sockets, err := custodianProcessSockets(pid)
	if err != nil {
		e.Logger.Warn("Failed collecting custodian child socket snapshot", "check_name", checkName, "pid", pid, "error", err)
		return
	}
	if len(sockets) == 0 {
		e.Logger.Debug("No custodian child sockets found", "check_name", checkName, "pid", pid)
		return
	}
	e.Logger.Info("Custodian child socket snapshot", "check_name", checkName, "pid", pid, "socket_count", len(sockets), "sockets", sockets)
}

type custodianLogTailCache struct {
	paths         []string
	nextDiscovery time.Time
}

func (e *CommandCustodianExecutor) logCustodianRunLogTail(outputDir, checkName, previous string, cache *custodianLogTailCache) string {
	logPaths, err := cachedCustodianRunLogPaths(outputDir, cache)
	if err != nil {
		e.Logger.Warn("Failed locating custodian run logs while process is still running", "check_name", checkName, "error", err)
		return previous
	}
	if len(logPaths) == 0 {
		return previous
	}
	_, logTail, err := readCustodianLogArtifactsForPaths(logPaths, custodianOutputTailBytes)
	if err != nil {
		e.Logger.Warn("Failed reading custodian run log tail while process is still running", "check_name", checkName, "error", err)
		return previous
	}
	if strings.TrimSpace(logTail) == "" || logTail == previous {
		return previous
	}
	e.Logger.Info("Custodian run log tail while process is still running", "check_name", checkName, "log_tail", logTail)
	return logTail
}

func cachedCustodianRunLogPaths(outputDir string, cache *custodianLogTailCache) ([]string, error) {
	if cache == nil {
		return findCustodianRunLogs(outputDir)
	}
	now := time.Now()
	if now.Before(cache.nextDiscovery) {
		return cache.paths, nil
	}
	logPaths, err := findCustodianRunLogs(outputDir)
	if err != nil {
		return nil, err
	}
	cache.paths = logPaths
	cache.nextDiscovery = now.Add(2 * time.Minute)
	if len(cache.paths) == 0 {
		cache.nextDiscovery = now.Add(30 * time.Second)
	}
	return cache.paths, nil
}

func custodianProcessSockets(pid int) ([]string, error) {
	if runtime.GOOS != "linux" {
		return []string{}, nil
	}
	if _, err := os.Stat("/proc"); err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			return []string{}, nil
		}
		return nil, err
	}
	inodes, err := processSocketInodes(pid)
	if err != nil {
		return nil, err
	}
	if len(inodes) == 0 {
		return []string{}, nil
	}
	sockets := make([]string, 0)
	tcp4, err := readProcNetTCP(filepath.Join("/proc", strconv.Itoa(pid), "net", "tcp"), inodes, false)
	if err != nil {
		return nil, err
	}
	sockets = append(sockets, tcp4...)
	tcp6, err := readProcNetTCP(filepath.Join("/proc", strconv.Itoa(pid), "net", "tcp6"), inodes, true)
	if err != nil {
		return nil, err
	}
	sockets = append(sockets, tcp6...)
	return sockets, nil
}

func processSocketInodes(pid int) (map[string]bool, error) {
	fdDir := filepath.Join("/proc", strconv.Itoa(pid), "fd")
	entries, err := os.ReadDir(fdDir)
	if err != nil {
		if errors.Is(err, fs.ErrNotExist) || errors.Is(err, os.ErrNotExist) {
			return map[string]bool{}, nil
		}
		return nil, err
	}
	inodes := map[string]bool{}
	for _, entry := range entries {
		target, err := os.Readlink(filepath.Join(fdDir, entry.Name()))
		if err != nil {
			if errors.Is(err, fs.ErrNotExist) || errors.Is(err, os.ErrNotExist) {
				continue
			}
			continue
		}
		if strings.HasPrefix(target, "socket:[") && strings.HasSuffix(target, "]") {
			inode := strings.TrimSuffix(strings.TrimPrefix(target, "socket:["), "]")
			inodes[inode] = true
		}
	}
	return inodes, nil
}

func readProcNetTCP(path string, inodes map[string]bool, ipv6 bool) ([]string, error) {
	content, err := os.ReadFile(path)
	if err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			return nil, nil
		}
		return nil, err
	}
	lines := strings.Split(string(content), "\n")
	sockets := make([]string, 0)
	for _, line := range lines[1:] {
		fields := strings.Fields(line)
		if len(fields) < 10 {
			continue
		}
		inode := fields[9]
		if !inodes[inode] {
			continue
		}
		local := decodeProcNetAddress(fields[1], ipv6)
		remote := decodeProcNetAddress(fields[2], ipv6)
		state := tcpStateName(fields[3])
		sockets = append(sockets, fmt.Sprintf("local=%s remote=%s state=%s inode=%s", local, remote, state, inode))
	}
	return sockets, nil
}

func decodeProcNetAddress(value string, ipv6 bool) string {
	parts := strings.Split(value, ":")
	if len(parts) != 2 {
		return value
	}
	port, err := strconv.ParseUint(parts[1], 16, 16)
	if err != nil {
		return value
	}
	if !ipv6 && len(parts[0]) == 8 {
		ipBytes, err := hex.DecodeString(parts[0])
		if err == nil && len(ipBytes) == 4 {
			return fmt.Sprintf("%d.%d.%d.%d:%d", ipBytes[3], ipBytes[2], ipBytes[1], ipBytes[0], port)
		}
	}
	if ipv6 && len(parts[0]) == 32 {
		ipBytes, err := hex.DecodeString(parts[0])
		if err == nil && len(ipBytes) == net.IPv6len {
			for index := 0; index < len(ipBytes); index += 4 {
				ipBytes[index], ipBytes[index+3] = ipBytes[index+3], ipBytes[index]
				ipBytes[index+1], ipBytes[index+2] = ipBytes[index+2], ipBytes[index+1]
			}
			return net.JoinHostPort(net.IP(ipBytes).String(), strconv.FormatUint(port, 10))
		}
	}
	return fmt.Sprintf("%s:%d", parts[0], port)
}

func tcpStateName(state string) string {
	switch strings.ToUpper(state) {
	case "01":
		return "ESTABLISHED"
	case "02":
		return "SYN_SENT"
	case "03":
		return "SYN_RECV"
	case "04":
		return "FIN_WAIT1"
	case "05":
		return "FIN_WAIT2"
	case "06":
		return "TIME_WAIT"
	case "07":
		return "CLOSE"
	case "08":
		return "CLOSE_WAIT"
	case "09":
		return "LAST_ACK"
	case "0A":
		return "LISTEN"
	case "0B":
		return "CLOSING"
	default:
		return state
	}
}

func readResourcesArtifact(outputDir string) (string, []interface{}, error) {
	resourcesPath, err := findResourcesJSON(outputDir)
	if err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			return "", []interface{}{}, nil
		}
		return "", nil, fmt.Errorf("failed to locate resources.json: %w", err)
	}

	content, err := os.ReadFile(resourcesPath)
	if err != nil {
		return resourcesPath, nil, fmt.Errorf("failed to read resources.json: %w", err)
	}

	if len(content) == 0 {
		return resourcesPath, []interface{}{}, nil
	}

	resources := make([]interface{}, 0)
	if err := json.Unmarshal(content, &resources); err != nil {
		return resourcesPath, nil, fmt.Errorf("failed to parse resources.json: %w", err)
	}

	return resourcesPath, resources, nil
}

func readCustodianLogArtifacts(outputDir string, maxBytes int) ([]string, string, error) {
	logPaths, err := findCustodianRunLogs(outputDir)
	if err != nil {
		return nil, "", err
	}
	return readCustodianLogArtifactsForPaths(logPaths, maxBytes)
}

func readCustodianLogArtifactsForPaths(logPaths []string, maxBytes int) ([]string, string, error) {
	if len(logPaths) == 0 {
		return nil, "", nil
	}

	sections := make([]string, 0, min(len(logPaths), custodianLogTailMaxSections))
	for i, logPath := range logPaths {
		content, err := readFileTail(logPath, maxBytes)
		if err != nil {
			return logPaths, "", fmt.Errorf("failed to read custodian log %s: %w", logPath, err)
		}
		if strings.TrimSpace(content) == "" {
			continue
		}
		if len(sections) >= custodianLogTailMaxSections {
			remaining := len(logPaths) - i
			sections = append(sections, fmt.Sprintf("custodian log tail truncated: %d additional custodian-run.log file(s) omitted", remaining))
			break
		}
		sections = append(sections, fmt.Sprintf("custodian log tail from %s:\n%s", logPath, content))
	}
	return logPaths, strings.Join(sections, "\n"), nil
}

func findCustodianRunLogs(outputDir string) ([]string, error) {
	logPaths := make([]string, 0)
	err := filepath.WalkDir(outputDir, func(path string, d fs.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if d.IsDir() || d.Name() != "custodian-run.log" {
			return nil
		}
		info, err := os.Lstat(path)
		if err != nil {
			return err
		}
		if info.Mode()&fs.ModeSymlink != 0 {
			return nil
		}
		logPaths = append(logPaths, path)
		return nil
	})
	if err != nil {
		return nil, fmt.Errorf("failed to locate custodian logs: %w", err)
	}
	slices.Sort(logPaths)
	return logPaths, nil
}

func readFileTail(path string, maxBytes int) (string, error) {
	file, err := os.Open(path)
	if err != nil {
		return "", err
	}
	defer file.Close()

	info, err := file.Stat()
	if err != nil {
		return "", err
	}
	size := info.Size()
	offset := int64(0)
	if maxBytes > 0 && size > int64(maxBytes) {
		offset = size - int64(maxBytes)
	}
	if _, err := file.Seek(offset, io.SeekStart); err != nil {
		return "", err
	}
	content, err := io.ReadAll(file)
	if err != nil {
		return "", err
	}
	return string(content), nil
}

func findResourcesJSON(outputDir string) (string, error) {
	found := ""
	err := filepath.WalkDir(outputDir, func(path string, d fs.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if d.IsDir() {
			return nil
		}
		if d.Name() == "resources.json" {
			found = path
			return filepath.SkipAll
		}
		return nil
	})

	if err != nil && !errors.Is(err, filepath.SkipAll) {
		return "", err
	}
	if found == "" {
		return "", fs.ErrNotExist
	}
	return found, nil
}

type StandardizedCheckInfo struct {
	Name     string                 `json:"name"`
	Resource string                 `json:"resource"`
	Provider string                 `json:"provider"`
	Index    int                    `json:"index"`
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

type StandardizedExecution struct {
	Status     string   `json:"status"`
	DryRun     bool     `json:"dry_run"`
	ExitCode   int      `json:"exit_code"`
	StartedAt  string   `json:"started_at"`
	EndedAt    string   `json:"ended_at"`
	DurationMS int64    `json:"duration_ms"`
	Stdout     string   `json:"stdout,omitempty"`
	Stderr     string   `json:"stderr,omitempty"`
	Error      string   `json:"error,omitempty"`
	Errors     []string `json:"errors,omitempty"`
	Warnings   []string `json:"warnings,omitempty"`
}

// StandardizedResourcePayload is the per-resource OPA input contract.
type StandardizedResourcePayload struct {
	SchemaVersion string                   `json:"schema_version"`
	Source        string                   `json:"source"`
	Check         StandardizedCheckInfo    `json:"check"`
	Resource      StandardizedResourceInfo `json:"resource"`
	Assessment    StandardizedAssessment   `json:"assessment"`
	Execution     StandardizedExecution    `json:"execution"`
	RawPolicy     map[string]interface{}   `json:"raw_policy"`
}

type StandardizedResourceInfo struct {
	ID             string            `json:"id"`
	Type           string            `json:"type"`
	Provider       string            `json:"provider"`
	AccountID      string            `json:"account_id,omitempty"`
	Region         string            `json:"region,omitempty"`
	IdentityFields map[string]string `json:"identity_fields,omitempty"`
	Data           interface{}       `json:"data"`
}

type StandardizedAssessment struct {
	Status               string `json:"status"`
	Matched              bool   `json:"matched"`
	InventoryStatus      string `json:"inventory_status"`
	MatchedResourceCount int    `json:"matched_resource_count"`
	ArtifactPath         string `json:"artifact_path,omitempty"`
	ResourcesPath        string `json:"resources_path,omitempty"`
}

type ResourceRecord struct {
	ID             string
	Type           string
	Provider       string
	AccountID      string
	Region         string
	IdentityFields map[string]string
	Data           interface{}
}

type InventoryBaseline struct {
	Execution    CustodianExecutionResult
	Records      []ResourceRecord
	ResourceType string
	Provider     string
	Err          error
}

func buildResourcePayload(
	check CustodianCheck,
	execution CustodianExecutionResult,
	record ResourceRecord,
	assessment StandardizedAssessment,
) *StandardizedResourcePayload {
	metadata := buildCheckMetadata(check)
	return &StandardizedResourcePayload{
		SchemaVersion: schemaVersionV2,
		Source:        sourceCloudCustodian,
		Check: StandardizedCheckInfo{
			Name:     check.Name,
			Resource: check.Resource,
			Provider: check.Provider,
			Index:    check.Index,
			Metadata: metadata,
		},
		Resource: StandardizedResourceInfo{
			ID:             record.ID,
			Type:           record.Type,
			Provider:       record.Provider,
			AccountID:      record.AccountID,
			Region:         record.Region,
			IdentityFields: record.IdentityFields,
			Data:           record.Data,
		},
		Assessment: assessment,
		Execution:  buildExecutionInfo(execution),
		RawPolicy:  check.RawPolicy,
	}
}

func buildCheckMetadata(check CustodianCheck) map[string]interface{} {
	var metadata map[string]interface{}
	for k, v := range check.RawPolicy {
		if k == "name" || k == "resource" {
			continue
		}
		if metadata == nil {
			metadata = map[string]interface{}{}
		}
		metadata[k] = v
	}
	return metadata
}

func buildExecutionInfo(execution CustodianExecutionResult) StandardizedExecution {
	status := "success"
	if execution.Error != "" {
		status = "error"
	}

	durationMS := int64(execution.EndedAt.Sub(execution.StartedAt) / time.Millisecond)
	if durationMS < 0 {
		durationMS = 0
	}

	var executionErrors []string
	if len(execution.Errors) > 0 {
		executionErrors = append([]string{}, execution.Errors...)
	}
	var executionWarnings []string
	if len(execution.DiagnosticWarnings) > 0 {
		executionWarnings = append([]string{}, execution.DiagnosticWarnings...)
	}

	return StandardizedExecution{
		Status:     status,
		DryRun:     true,
		ExitCode:   execution.ExitCode,
		StartedAt:  execution.StartedAt.UTC().Format(time.RFC3339Nano),
		EndedAt:    execution.EndedAt.UTC().Format(time.RFC3339Nano),
		DurationMS: durationMS,
		Stdout:     execution.Stdout,
		Stderr:     execution.Stderr,
		Error:      execution.Error,
		Errors:     executionErrors,
		Warnings:   executionWarnings,
	}
}

func resourceRecordDisambiguator(record ResourceRecord) string {
	identityFields := map[string]string{}
	for key, value := range record.IdentityFields {
		if key == "resource_hash" || strings.TrimSpace(value) == "" || value == record.ID {
			continue
		}
		identityFields[key] = value
	}
	if len(identityFields) > 0 {
		return stableHashValue(identityFields)
	}
	return hashResource(record.Data)
}

func resourceIDCollisions(records []ResourceRecord) map[string]bool {
	counts := map[string]int{}
	for _, record := range records {
		counts[record.ID]++
	}
	collisions := map[string]bool{}
	for resourceID, count := range counts {
		if count > 1 {
			collisions[resourceID] = true
		}
	}
	return collisions
}

func mergeCollisionIDs(groups ...map[string]bool) map[string]bool {
	merged := map[string]bool{}
	for _, group := range groups {
		for resourceID := range group {
			merged[resourceID] = true
		}
	}
	return merged
}

func disambiguateResourceRecords(records []ResourceRecord, collisionIDs map[string]bool) (map[string]ResourceRecord, int) {
	result := make(map[string]ResourceRecord, len(records))
	collisionCount := 0
	grouped := map[string][]ResourceRecord{}
	for _, record := range records {
		grouped[record.ID] = append(grouped[record.ID], record)
	}
	for resourceID, group := range grouped {
		if !collisionIDs[resourceID] {
			result[resourceID] = group[0]
			continue
		}
		collisionCount += len(group) - 1
		for _, record := range group {
			disambiguated := record
			if disambiguated.IdentityFields == nil {
				disambiguated.IdentityFields = map[string]string{}
			}
			hash := hashResource(disambiguated.Data)
			disambiguated.IdentityFields["resource_hash"] = hash
			baseSuffix := resourceRecordDisambiguator(disambiguated)
			suffix := hash
			if baseSuffix != "" && baseSuffix != hash {
				suffix = fmt.Sprintf("%s-%s", baseSuffix, hash)
			}
			disambiguatedID := fmt.Sprintf("%s#%s", resourceID, suffix)
			for i := 2; ; i++ {
				if _, exists := result[disambiguatedID]; !exists {
					break
				}
				disambiguatedID = fmt.Sprintf("%s#%s-%d", resourceID, suffix, i)
			}
			disambiguated.ID = disambiguatedID
			result[disambiguatedID] = disambiguated
		}
	}

	return result, collisionCount
}

func normalizeForHash(value interface{}) interface{} {
	switch typed := value.(type) {
	case map[string]interface{}:
		normalized := make(map[string]interface{}, len(typed))
		for key, nested := range typed {
			normalized[key] = normalizeForHash(nested)
		}
		return normalized
	case map[interface{}]interface{}:
		normalized := make(map[string]interface{}, len(typed))
		keys := make([]string, 0, len(typed))
		keyedValues := make(map[string]interface{}, len(typed))
		for key, nested := range typed {
			stringKey := fmt.Sprint(key)
			keys = append(keys, stringKey)
			keyedValues[stringKey] = normalizeForHash(nested)
		}
		slices.Sort(keys)
		for _, key := range keys {
			normalized[key] = keyedValues[key]
		}
		return normalized
	case []interface{}:
		normalized := make([]interface{}, 0, len(typed))
		for _, item := range typed {
			normalized = append(normalized, normalizeForHash(item))
		}
		return normalized
	}

	rv := reflect.ValueOf(value)
	if !rv.IsValid() {
		return nil
	}
	switch rv.Kind() {
	case reflect.Map:
		keys := rv.MapKeys()
		stringKeys := make([]string, 0, len(keys))
		keyedValues := make(map[string]interface{}, len(keys))
		for _, key := range keys {
			stringKey := fmt.Sprint(key.Interface())
			stringKeys = append(stringKeys, stringKey)
			keyedValues[stringKey] = normalizeForHash(rv.MapIndex(key).Interface())
		}
		slices.Sort(stringKeys)
		normalized := make(map[string]interface{}, len(keys))
		for _, key := range stringKeys {
			normalized[key] = keyedValues[key]
		}
		return normalized
	case reflect.Slice, reflect.Array:
		length := rv.Len()
		normalized := make([]interface{}, 0, length)
		for i := 0; i < length; i++ {
			normalized = append(normalized, normalizeForHash(rv.Index(i).Interface()))
		}
		return normalized
	default:
		return value
	}
}

func stableHashValue(value interface{}) string {
	content, err := json.Marshal(normalizeForHash(value))
	if err != nil {
		content = []byte(fmt.Sprintf("%T:%v", value, value))
	}
	sum := sha256.Sum256(content)
	return hex.EncodeToString(sum[:])
}

func (p *CloudCustodianPlugin) buildResourceRecord(resourceType string, resource interface{}) ResourceRecord {
	provider := extractProvider(resourceType)
	identityFields := map[string]string{}
	fieldPaths := p.identityFieldPaths(resourceType)

	resourceID := ""
	for _, fieldPath := range fieldPaths {
		value, ok := resourceStringAtPath(resource, fieldPath)
		if !ok || value == "" {
			continue
		}
		identityFields[fieldPath] = value
		if resourceID == "" {
			resourceID = value
		}
	}
	if resourceID == "" {
		resourceID = hashResource(resource)
		identityFields["resource_hash"] = resourceID
	}
	resourceID = canonicalResourceID(resourceType, provider, resourceID)
	if strings.HasPrefix(resourceID, "arn:") {
		identityFields["arn"] = resourceID
	}

	record := ResourceRecord{
		ID:             resourceID,
		Type:           resourceType,
		Provider:       provider,
		IdentityFields: identityFields,
		Data:           resource,
	}
	if accountID, ok := firstResourceString(resource, []string{"AccountId", "AccountID", "account_id", "accountId", "OwnerId", "OwnerID", "owner_id", "c7n:account-id"}); ok {
		record.AccountID = accountID
	}
	if region, ok := firstResourceString(resource, []string{"Region", "region", "AwsRegion", "aws_region", "awsRegion", "c7n:region"}); ok {
		record.Region = region
	}
	return record
}

func canonicalResourceID(resourceType string, provider string, resourceID string) string {
	resourceID = strings.TrimSpace(resourceID)
	if resourceID == "" || strings.HasPrefix(resourceID, "arn:") {
		return resourceID
	}
	if provider == "aws" && resourceType == "aws.hostedzone" {
		hostedZoneID := strings.TrimPrefix(resourceID, "/")
		if strings.HasPrefix(hostedZoneID, "hostedzone/") {
			return "arn:aws:route53:::" + hostedZoneID
		}
		if strings.HasPrefix(hostedZoneID, "Z") {
			return "arn:aws:route53:::hostedzone/" + hostedZoneID
		}
	}
	return resourceID
}

func (p *CloudCustodianPlugin) identityFieldPaths(resourceType string) []string {
	paths := make([]string, 0)
	if p.parsedConfig != nil {
		if configured, ok := p.parsedConfig.ResourceIdentityFields[resourceType]; ok {
			paths = append(paths, configured...)
		}
	}
	paths = append(paths,
		"Arn",
		"ARN",
		"arn",
		"Id",
		"ID",
		"id",
		"InstanceId",
		"instance_id",
		"InstanceID",
		"ResourceId",
		"resource_id",
		"Name",
		"name",
	)
	return compactUniqueStrings(paths)
}

func compactUniqueStrings(values []string) []string {
	seen := map[string]bool{}
	result := make([]string, 0, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" || seen[value] {
			continue
		}
		seen[value] = true
		result = append(result, value)
	}
	return result
}

func firstResourceString(resource interface{}, fieldPaths []string) (string, bool) {
	for _, fieldPath := range fieldPaths {
		value, ok := resourceStringAtPath(resource, fieldPath)
		if ok && value != "" {
			return value, true
		}
	}
	return "", false
}

func resourceStringAtPath(resource interface{}, fieldPath string) (string, bool) {
	current := resource
	for _, part := range strings.Split(fieldPath, ".") {
		part = strings.TrimSpace(part)
		if part == "" {
			return "", false
		}
		switch typed := current.(type) {
		case map[string]interface{}:
			value, ok := typed[part]
			if !ok {
				return "", false
			}
			current = value
		case map[interface{}]interface{}:
			value, ok := typed[part]
			if !ok {
				return "", false
			}
			current = value
		default:
			return "", false
		}
	}

	switch value := current.(type) {
	case string:
		return strings.TrimSpace(value), strings.TrimSpace(value) != ""
	case json.Number:
		return value.String(), value.String() != ""
	case float64:
		return strconv.FormatFloat(value, 'f', -1, 64), true
	case float32:
		return strconv.FormatFloat(float64(value), 'f', -1, 32), true
	case int:
		return strconv.Itoa(value), true
	case int64:
		return strconv.FormatInt(value, 10), true
	case bool:
		return strconv.FormatBool(value), true
	default:
		return "", false
	}
}

func hashResource(resource interface{}) string {
	normalized := normalizeForHash(resource)
	content, err := json.Marshal(normalized)
	if err != nil {
		content = []byte(fmt.Sprintf("%#v", normalized))
	}
	sum := sha256.Sum256(content)
	return hex.EncodeToString(sum[:])
}

func buildInventoryCheck(resourceType string) CustodianCheck {
	provider := extractProvider(resourceType)
	name := fmt.Sprintf("inventory-%s", sanitizeIdentifier(resourceType))
	return CustodianCheck{
		Index:    -1,
		Name:     name,
		Resource: resourceType,
		Provider: provider,
		RawPolicy: map[string]interface{}{
			"name":     name,
			"resource": resourceType,
		},
	}
}

func custodianPolicyForExecution(policy map[string]interface{}) map[string]interface{} {
	if policy == nil {
		return nil
	}

	out := make(map[string]interface{}, len(policy))
	for k, v := range policy {
		if k == nonComplianceMessageField {
			continue
		}
		out[k] = v
	}
	return out
}

func parseCustodianChecks(policyYAML []byte) ([]CustodianCheck, error) {
	decoded := map[string]interface{}{}
	if err := yaml.Unmarshal(policyYAML, &decoded); err != nil {
		return nil, fmt.Errorf("failed to parse cloud custodian policies yaml: %w", err)
	}

	rawPolicies, ok := decoded["policies"]
	if !ok {
		return nil, errors.New("policy document must contain top-level policies array")
	}

	policies, ok := rawPolicies.([]interface{})
	if !ok {
		return nil, errors.New("top-level policies must be an array")
	}
	if len(policies) == 0 {
		return nil, errors.New("top-level policies array must not be empty")
	}

	checks := make([]CustodianCheck, 0, len(policies))
	for idx, raw := range policies {
		normalized := normalizeYAMLValue(raw)
		policyMap, mapOK := normalized.(map[string]interface{})
		if !mapOK {
			checks = append(checks, CustodianCheck{
				Index:       idx,
				Name:        fmt.Sprintf("policy-%d", idx+1),
				Resource:    "unknown",
				Provider:    "unknown",
				RawPolicy:   map[string]interface{}{"value": normalized},
				ParseErrors: []string{fmt.Sprintf("policy entry at index %d is not an object", idx)},
			})
			continue
		}

		name := strings.TrimSpace(asString(policyMap["name"]))
		resource := strings.TrimSpace(asString(policyMap["resource"]))
		parseIssues := make([]string, 0)
		if name == "" {
			name = fmt.Sprintf("policy-%d", idx+1)
			parseIssues = append(parseIssues, fmt.Sprintf("policy entry at index %d missing required name", idx))
		}
		if resource == "" {
			resource = "unknown"
			parseIssues = append(parseIssues, fmt.Sprintf("policy %q missing required resource", name))
		}

		checks = append(checks, CustodianCheck{
			Index:       idx,
			Name:        name,
			Resource:    resource,
			Provider:    extractProvider(resource),
			RawPolicy:   policyMap,
			ParseErrors: parseIssues,
		})
	}

	return checks, nil
}

func normalizeYAMLValue(in interface{}) interface{} {
	switch value := in.(type) {
	case map[string]interface{}:
		out := make(map[string]interface{}, len(value))
		for k, v := range value {
			out[k] = normalizeYAMLValue(v)
		}
		return out
	case map[interface{}]interface{}:
		out := make(map[string]interface{}, len(value))
		for k, v := range value {
			out[fmt.Sprint(k)] = normalizeYAMLValue(v)
		}
		return out
	case []interface{}:
		out := make([]interface{}, len(value))
		for i, v := range value {
			out[i] = normalizeYAMLValue(v)
		}
		return out
	default:
		return value
	}
}

func asString(value interface{}) string {
	stringValue, ok := value.(string)
	if !ok {
		return ""
	}
	return stringValue
}

func extractProvider(resource string) string {
	resource = strings.TrimSpace(resource)
	if resource == "" {
		return "unknown"
	}

	parts := strings.SplitN(resource, ".", 2)
	if len(parts) < 2 {
		return "unknown"
	}
	provider := strings.TrimSpace(parts[0])
	if provider == "" {
		return "unknown"
	}
	return provider
}

func resolvePoliciesYAML(ctx context.Context, inlineYAML string, policiesPath string) ([]byte, error) {
	if strings.TrimSpace(inlineYAML) != "" {
		return []byte(inlineYAML), nil
	}

	pathValue := strings.TrimSpace(policiesPath)
	if pathValue == "" {
		return nil, errors.New("policies path is required when policies_yaml is empty")
	}

	// Treat values that do not look like URLs as local filesystem paths.
	// This handles Windows paths like `C:\policies.yaml` correctly.
	if !strings.Contains(pathValue, "://") {
		content, err := os.ReadFile(pathValue)
		if err != nil {
			return nil, fmt.Errorf("failed to read local policies file: %w", err)
		}
		return content, nil
	}

	parsedURL, err := url.Parse(pathValue)
	if err != nil {
		return nil, fmt.Errorf("invalid policies_path: %w", err)
	}

	switch parsedURL.Scheme {
	case "":
		content, err := os.ReadFile(pathValue)
		if err != nil {
			return nil, fmt.Errorf("failed to read local policies file: %w", err)
		}
		return content, nil
	case "file":
		if parsedURL.Path == "" {
			return nil, errors.New("file:// policies_path must include a file path")
		}
		content, err := os.ReadFile(parsedURL.Path)
		if err != nil {
			return nil, fmt.Errorf("failed to read file:// policies file: %w", err)
		}
		return content, nil
	case "http", "https":
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, pathValue, nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create request for policies_path: %w", err)
		}
		httpClient := &http.Client{Timeout: defaultRemotePolicyTimeout}
		resp, err := httpClient.Do(req)
		if err != nil {
			return nil, fmt.Errorf("failed to fetch policies_path: %w", err)
		}
		defer resp.Body.Close()

		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
			return nil, fmt.Errorf("unexpected status code %d while fetching policies_path", resp.StatusCode)
		}
		if resp.ContentLength > defaultMaxRemotePolicyBytes {
			return nil, fmt.Errorf("policies_path response too large: content-length=%d exceeds max=%d bytes", resp.ContentLength, defaultMaxRemotePolicyBytes)
		}

		content, err := io.ReadAll(io.LimitReader(resp.Body, defaultMaxRemotePolicyBytes+1))
		if err != nil {
			return nil, fmt.Errorf("failed to read policies_path response body: %w", err)
		}
		if len(content) > defaultMaxRemotePolicyBytes {
			return nil, fmt.Errorf("policies_path response too large: size=%d exceeds max=%d bytes", len(content), defaultMaxRemotePolicyBytes)
		}
		return content, nil
	default:
		return nil, fmt.Errorf("unsupported policies_path scheme: %s", parsedURL.Scheme)
	}
}

// PolicyEvaluator wraps OPA policy execution so eval loop behavior can be tested with mocks.
type PolicyEvaluator interface {
	Generate(
		ctx context.Context,
		policyPath string,
		labels map[string]string,
		subjects []*proto.Subject,
		components []*proto.Component,
		inventory []*proto.InventoryItem,
		actors []*proto.OriginActor,
		activities []*proto.Activity,
		data interface{},
	) ([]*proto.Evidence, error)
}

type DefaultPolicyEvaluator struct {
	Logger hclog.Logger
}

func (e *DefaultPolicyEvaluator) Generate(
	ctx context.Context,
	policyPath string,
	labels map[string]string,
	subjects []*proto.Subject,
	components []*proto.Component,
	inventory []*proto.InventoryItem,
	actors []*proto.OriginActor,
	activities []*proto.Activity,
	data interface{},
) ([]*proto.Evidence, error) {
	e.Logger.Debug("Evaluating OPA policy against check payload", "policy_path", policyPath, "labels", labels)
	processor := policyManager.NewPolicyProcessor(
		e.Logger,
		labels,
		subjects,
		components,
		inventory,
		actors,
		activities,
	)
	evidence, err := processor.GenerateResults(ctx, policyPath, data)
	if err != nil {
		e.Logger.Warn("OPA policy evaluation failed", "policy_path", policyPath, "error", err)
		return evidence, err
	}
	e.Logger.Debug("OPA policy evaluation succeeded", "policy_path", policyPath, "evidence_count", len(evidence))
	return evidence, nil
}

type CloudCustodianPlugin struct {
	Logger hclog.Logger

	config       *PluginConfig
	parsedConfig *ParsedConfig
	checks       []CustodianCheck

	executor  CustodianExecutor
	evaluator PolicyEvaluator
}

func (p *CloudCustodianPlugin) Configure(req *proto.ConfigureRequest) (*proto.ConfigureResponse, error) {
	p.Logger.Debug("Received raw plugin configuration", "config_keys", sortedKeys(req.Config))

	config := &PluginConfig{}
	if err := mapstructure.Decode(req.Config, config); err != nil {
		p.Logger.Error("Error decoding config", "error", err)
		return nil, err
	}

	parsed, err := config.Parse()
	if err != nil {
		p.Logger.Error("Error parsing config", "error", err)
		return nil, err
	}
	p.Logger.Debug("Parsed plugin configuration",
		"has_inline_policies_yaml", strings.TrimSpace(parsed.PoliciesYAML) != "",
		"policies_path", parsed.PoliciesPath,
		"custodian_binary", parsed.CustodianBinary,
		"aws_regions", parsed.AWSRegions,
		"check_timeout", parsed.CheckTimeout.String(),
		"policy_labels", parsed.PolicyLabels,
		"debug_dump_payloads", parsed.DebugDumpPayloads,
		"debug_payload_output_dir", parsed.DebugPayloadOutputDir,
		"custodian_aws_api_trace", parsed.CustodianAWSAPITrace,
		"custodian_network_diagnostics", parsed.CustodianNetworkDiag,
		"custodian_network_diagnostic_endpoints", parsed.CustodianNetworkDiagnosticEndpoints,
		"custodian_log_tail_during_run", parsed.CustodianLogTail,
		"preserve_execution_artifacts", parsed.PreserveArtifacts,
	)

	resolvedPolicies, err := resolvePoliciesYAML(context.Background(), parsed.PoliciesYAML, parsed.PoliciesPath)
	if err != nil {
		p.Logger.Error("Error loading cloud custodian policies", "error", err)
		return nil, err
	}
	p.Logger.Debug("Resolved cloud custodian policy source", "policy_yaml_bytes", len(resolvedPolicies))

	checks, err := parseCustodianChecks(resolvedPolicies)
	if err != nil {
		p.Logger.Error("Error parsing cloud custodian policies", "error", err)
		return nil, err
	}
	parseErrorChecks := 0
	for _, check := range checks {
		if len(check.ParseErrors) > 0 {
			parseErrorChecks++
			p.Logger.Debug("Parsed check with non-fatal parse issues", "check_name", check.Name, "index", check.Index, "parse_errors", check.ParseErrors)
		}
	}

	parsed.PoliciesYAML = string(resolvedPolicies)
	p.config = config
	p.parsedConfig = parsed
	p.checks = checks

	if parsed.DebugDumpPayloads {
		if err := os.MkdirAll(parsed.DebugPayloadOutputDir, 0o755); err != nil {
			p.Logger.Error("Failed creating debug payload output directory", "debug_payload_output_dir", parsed.DebugPayloadOutputDir, "error", err)
			return nil, fmt.Errorf("failed creating debug payload output directory %q: %w", parsed.DebugPayloadOutputDir, err)
		}
		p.Logger.Debug("Debug payload dumping enabled", "debug_payload_output_dir", parsed.DebugPayloadOutputDir)
	}

	if p.executor == nil {
		p.executor = &CommandCustodianExecutor{Logger: p.Logger.Named("custodian-executor")}
	}
	if p.evaluator == nil {
		p.evaluator = &DefaultPolicyEvaluator{Logger: p.Logger.Named("policy-evaluator")}
	}

	p.Logger.Info("Cloud Custodian Plugin configured", "checks", len(checks), "checks_with_parse_errors", parseErrorChecks)
	return &proto.ConfigureResponse{}, nil
}

func (p *CloudCustodianPlugin) Init(req *proto.InitRequest, apiHelper runner.ApiHelper) (*proto.InitResponse, error) {
	p.Logger.Debug("Cloud Custodian Plugin Init called",
		"configured", p.parsedConfig != nil,
		"policy_paths", req.GetPolicyPaths(),
		"policy_paths_count", len(req.GetPolicyPaths()),
		"checks_count", len(p.checks),
	)
	if p.parsedConfig == nil {
		p.Logger.Error("Cloud Custodian Plugin Init failed because plugin is not configured")
		return nil, errors.New("plugin not configured")
	}

	resourceTypes := p.uniqueResourceTypes()
	subjectTemplates := p.buildSubjectTemplates(resourceTypes)
	templateNames := make([]string, 0, len(subjectTemplates))
	for _, subjectTemplate := range subjectTemplates {
		templateNames = append(templateNames, subjectTemplate.GetName())
	}
	p.Logger.Debug("Cloud Custodian Plugin Init prepared subject templates",
		"resource_types", resourceTypes,
		"subject_template_count", len(subjectTemplates),
		"subject_template_names", templateNames,
	)

	p.Logger.Debug("Cloud Custodian Plugin Init delegating subject and risk template upsert",
		"policy_paths", req.GetPolicyPaths(),
		"subject_template_count", len(subjectTemplates),
	)
	resp, err := runner.InitWithSubjectsAndRisksFromPolicies(
		context.Background(),
		p.Logger,
		req,
		apiHelper,
		subjectTemplates,
	)
	if err != nil {
		p.Logger.Error("Cloud Custodian Plugin Init failed while upserting subject or risk templates", "error", err)
		return resp, err
	}
	p.Logger.Debug("Cloud Custodian Plugin Init completed",
		"subject_template_count", len(subjectTemplates),
		"policy_paths_count", len(req.GetPolicyPaths()),
	)
	return resp, nil
}

func (p *CloudCustodianPlugin) buildSubjectTemplates(resourceTypes []string) []*proto.SubjectTemplate {
	templates := make([]*proto.SubjectTemplate, 0, len(resourceTypes))
	for _, resourceType := range resourceTypes {
		provider := extractProvider(resourceType)
		templates = append(templates, &proto.SubjectTemplate{
			Name: fmt.Sprintf("cloud-custodian-%s", sanitizeIdentifier(resourceType)),
			// These templates represent cloud resources collected during evaluation.
			Type:                proto.SubjectType_SUBJECT_TYPE_RESOURCE,
			TitleTemplate:       "Cloud Resource: {{ .resource_type }} {{ .resource_id }}",
			DescriptionTemplate: "Cloud Custodian resource {{ .resource_id }} of type {{ .resource_type }} from provider {{ .provider }}",
			PurposeTemplate:     "Represents a cloud resource collected by Cloud Custodian for compliance evaluation.",
			IdentityLabelKeys:   []string{"provider", "resource_type", "resource_id"},
			Props: []*proto.SubjectProp{
				{Name: "provider", Value: provider},
				{Name: "resource_type", Value: resourceType},
			},
			SelectorLabels: []*proto.SubjectLabelSelector{
				{Key: "source", Value: sourceCloudCustodian},
				{Key: "resource_type", Value: resourceType},
			},
			LabelSchema: []*proto.SubjectLabelSchema{
				{Key: "provider", Description: "Cloud provider derived from the Cloud Custodian resource type."},
				{Key: "resource_type", Description: "Cloud Custodian resource type such as aws.ec2 or aws.s3."},
				{Key: "resource_id", Description: "Stable resource identifier extracted from the resource data."},
				{Key: "account_id", Description: "Cloud account identifier when available in the resource data."},
				{Key: "region", Description: "Cloud region when available in the resource data."},
			},
		})
	}
	return templates
}

func (p *CloudCustodianPlugin) uniqueResourceTypes() []string {
	seen := map[string]bool{}
	resourceTypes := make([]string, 0)
	for _, check := range p.checks {
		resourceType := strings.TrimSpace(check.Resource)
		if resourceType == "" || resourceType == "unknown" || len(check.ParseErrors) > 0 || seen[resourceType] {
			continue
		}
		seen[resourceType] = true
		resourceTypes = append(resourceTypes, resourceType)
	}
	slices.Sort(resourceTypes)
	return resourceTypes
}

func (p *CloudCustodianPlugin) Eval(req *proto.EvalRequest, apiHelper runner.ApiHelper) (*proto.EvalResponse, error) {
	ctx := context.Background()

	if p.parsedConfig == nil {
		return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, errors.New("plugin not configured")
	}
	if len(p.checks) == 0 {
		return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, errors.New("no cloud custodian checks configured")
	}
	if len(req.GetPolicyPaths()) == 0 {
		return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, errors.New("no policy paths provided")
	}

	executionRoot, err := os.MkdirTemp("", "ccf-cloud-custodian-*")
	if err != nil {
		return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, fmt.Errorf("failed to create execution workspace: %w", err)
	}
	pendingEvidences := make([]*proto.Evidence, 0, evidenceBatchSize)
	totalEvidenceCount := 0
	var accumulatedErrors error
	successfulPolicyRuns := 0
	hadCheckExecutionFailures := false
	defer func() {
		if p.parsedConfig.PreserveArtifacts && hadCheckExecutionFailures {
			p.Logger.Warn("Preserving cloud custodian execution artifacts after check execution failure", "execution_root", executionRoot)
			return
		}
		if err := os.RemoveAll(executionRoot); err != nil {
			p.Logger.Warn("Failed removing temporary execution root", "execution_root", executionRoot, "error", err)
		}
	}()
	p.Logger.Debug("Created temporary execution root", "execution_root", executionRoot)

	baselines := p.collectInventoryBaselines(ctx, executionRoot)
	for resourceType, baseline := range baselines {
		if baseline == nil || baseline.Err != nil || len(baseline.Execution.DiagnosticWarnings) == 0 {
			continue
		}
		err := formatExecutionDiagnosticWarnings(baseline.Execution.DiagnosticWarnings)
		p.Logger.Warn("Inventory baseline completed with unavailable AWS service endpoints", "resource", resourceType, "error", err)
		accumulatedErrors = errors.Join(accumulatedErrors, err)
		hadCheckExecutionFailures = true
	}
	for _, check := range p.checks {
		p.Logger.Debug("Processing check", "check_name", check.Name, "check_index", check.Index, "resource", check.Resource, "provider", check.Provider)
		if len(check.ParseErrors) > 0 {
			p.Logger.Warn("Skipping custodian execution due to check parse issues", "check_name", check.Name, "parse_errors", check.ParseErrors)
			accumulatedErrors = errors.Join(accumulatedErrors, fmt.Errorf("check %s has parse errors: %s", check.Name, strings.Join(check.ParseErrors, "; ")))
			hadCheckExecutionFailures = true
			continue
		}

		baseline := baselines[check.Resource]
		if baseline == nil || baseline.Err != nil {
			var err error
			if baseline != nil && baseline.Err != nil {
				err = fmt.Errorf("inventory baseline unavailable for resource type %s: %w", check.Resource, baseline.Err)
			} else {
				err = fmt.Errorf("inventory baseline unavailable for resource type %s", check.Resource)
			}
			p.Logger.Error("Skipping check due to unavailable inventory baseline", "check_name", check.Name, "resource", check.Resource, "error", err)
			accumulatedErrors = errors.Join(accumulatedErrors, err)
			hadCheckExecutionFailures = true
			continue
		}

		checkDir := filepath.Join(executionRoot, fmt.Sprintf("%03d-%s", check.Index+1, sanitizeIdentifier(check.Name)))
		execution := p.executor.Execute(ctx, CustodianExecutionRequest{
			BinaryPath:                 p.parsedConfig.CustodianBinary,
			Check:                      check,
			Timeout:                    p.parsedConfig.CheckTimeout,
			OutputDir:                  checkDir,
			Debug:                      p.parsedConfig.CustodianDebug,
			Verbose:                    p.parsedConfig.CustodianVerbose,
			AWSRegions:                 p.parsedConfig.AWSRegions,
			AWSAPITrace:                p.parsedConfig.CustodianAWSAPITrace,
			NetworkDiagnostics:         p.parsedConfig.CustodianNetworkDiag,
			NetworkDiagnosticEndpoints: p.parsedConfig.CustodianNetworkDiagnosticEndpoints,
			LogTailDuringRun:           p.parsedConfig.CustodianLogTail,
		})
		if execution.Err != nil || execution.Error != "" {
			err := formatExecutionFailure(check.Name, execution)
			p.Logger.Error("Skipping resource evaluation due to check execution error", "check_name", check.Name, "error", err)
			accumulatedErrors = errors.Join(accumulatedErrors, err)
			hadCheckExecutionFailures = true
			continue
		}

		payloads := p.buildResourcePayloadsForCheck(check, execution, baseline)
		payloadStats := summarizePayloadAssessments(payloads)
		p.Logger.Debug("Built standardized resource payloads",
			"check_name", check.Name,
			"payload_count", len(payloads),
			"matched_resource_count", len(execution.Resources),
			"baseline_resource_count", len(baseline.Records),
			"compliant_resource_count", payloadStats.Compliant,
			"non_compliant_resource_count", payloadStats.NonCompliant,
			"missing_from_baseline_count", payloadStats.MissingFromBaseline,
		)
		if len(baseline.Records) == 0 && len(execution.Resources) > 0 {
			p.Logger.Warn("No compliant resource payloads can be generated because inventory baseline is empty while policy returned matched resources",
				"check_name", check.Name,
				"resource", check.Resource,
				"matched_resource_count", len(execution.Resources),
				"resources_path", execution.ResourcesPath,
			)
		}
		if p.parsedConfig.DebugDumpPayloads {
			for _, payload := range payloads {
				if err := p.dumpStandardizedPayload(payload); err != nil {
					p.Logger.Warn("Failed writing debug standardized payload", "check_name", payload.Check.Name, "resource_id", payload.Resource.ID, "error", err)
				}
			}
		}

		for _, payload := range payloads {
			evidences, evalErr, successfulRuns := p.evaluateResourcePolicies(ctx, payload, req.GetPolicyPaths())
			pendingEvidences = append(pendingEvidences, evidences...)
			totalEvidenceCount += len(evidences)
			successfulPolicyRuns += successfulRuns
			p.Logger.Debug("Completed policy evaluations for resource",
				"check_name", payload.Check.Name,
				"resource_id", payload.Resource.ID,
				"assessment_status", payload.Assessment.Status,
				"successful_policy_runs", successfulRuns,
				"produced_evidence_count", len(evidences),
				"had_eval_error", evalErr != nil,
			)
			if evalErr != nil {
				accumulatedErrors = errors.Join(accumulatedErrors, evalErr)
			}
			for len(pendingEvidences) >= evidenceBatchSize {
				if err := p.submitEvidenceBatch(ctx, apiHelper, pendingEvidences[:evidenceBatchSize]); err != nil {
					return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, err
				}
				pendingEvidences = pendingEvidences[evidenceBatchSize:]
			}
		}
		if len(execution.DiagnosticWarnings) > 0 {
			err := formatExecutionDiagnosticWarnings(execution.DiagnosticWarnings)
			p.Logger.Warn("Check completed with unavailable AWS service endpoints", "check_name", check.Name, "error", err)
			accumulatedErrors = errors.Join(accumulatedErrors, err)
			hadCheckExecutionFailures = true
		}
	}

	if len(pendingEvidences) > 0 {
		for len(pendingEvidences) > 0 {
			batch := pendingEvidences
			if len(batch) > evidenceBatchSize {
				batch = batch[:evidenceBatchSize]
			}
			if err := p.submitEvidenceBatch(ctx, apiHelper, batch); err != nil {
				return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, err
			}
			pendingEvidences = pendingEvidences[len(batch):]
			if len(pendingEvidences) == 0 {
				pendingEvidences = nil
			}
		}
	} else {
		p.Logger.Warn("No evidence generated by current evaluation run")
	}

	if successfulPolicyRuns == 0 && totalEvidenceCount == 0 {
		if accumulatedErrors == nil {
			accumulatedErrors = errors.New("policy evaluation failed for all checks")
		}
		return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, accumulatedErrors
	}
	if hadCheckExecutionFailures {
		if accumulatedErrors == nil {
			accumulatedErrors = errors.New("one or more cloud custodian checks failed to execute")
		}
		return &proto.EvalResponse{Status: proto.ExecutionStatus_FAILURE}, accumulatedErrors
	}

	if accumulatedErrors != nil {
		p.Logger.Warn("Completed with non-fatal policy evaluation errors", "error", accumulatedErrors)
	}

	return &proto.EvalResponse{Status: proto.ExecutionStatus_SUCCESS}, nil
}

func (p *CloudCustodianPlugin) submitEvidenceBatch(ctx context.Context, apiHelper runner.ApiHelper, evidences []*proto.Evidence) error {
	p.Logger.Debug("Submitting evidence batch via ApiHelper", "evidence_count", len(evidences))
	if err := apiHelper.CreateEvidence(ctx, evidences); err != nil {
		p.Logger.Error("Error creating evidence", "error", err)
		return err
	}
	return nil
}

func (p *CloudCustodianPlugin) collectInventoryBaselines(ctx context.Context, executionRoot string) map[string]*InventoryBaseline {
	baselines := map[string]*InventoryBaseline{}
	for _, resourceType := range p.uniqueResourceTypes() {
		check := buildInventoryCheck(resourceType)
		outputDir := filepath.Join(executionRoot, fmt.Sprintf("inventory-%s", sanitizeIdentifier(resourceType)))
		execution := p.executor.Execute(ctx, CustodianExecutionRequest{
			BinaryPath:                 p.parsedConfig.CustodianBinary,
			Check:                      check,
			Timeout:                    p.parsedConfig.CheckTimeout,
			OutputDir:                  outputDir,
			Debug:                      p.parsedConfig.CustodianDebug,
			Verbose:                    p.parsedConfig.CustodianVerbose,
			AWSRegions:                 p.parsedConfig.AWSRegions,
			AWSAPITrace:                p.parsedConfig.CustodianAWSAPITrace,
			NetworkDiagnostics:         p.parsedConfig.CustodianNetworkDiag,
			NetworkDiagnosticEndpoints: p.parsedConfig.CustodianNetworkDiagnosticEndpoints,
			LogTailDuringRun:           p.parsedConfig.CustodianLogTail,
		})

		var baselineErr error
		if execution.Err != nil || execution.Error != "" {
			baselineErr = formatExecutionFailure(check.Name, execution)
		}
		records := make([]ResourceRecord, 0, len(execution.Resources))
		for _, resource := range execution.Resources {
			records = append(records, p.buildResourceRecord(resourceType, resource))
		}
		collisionIDs := resourceIDCollisions(records)
		_, collisionCount := disambiguateResourceRecords(records, collisionIDs)
		baseline := &InventoryBaseline{
			Execution:    execution,
			Records:      records,
			ResourceType: resourceType,
			Provider:     extractProvider(resourceType),
			Err:          baselineErr,
		}
		baselines[resourceType] = baseline
		p.Logger.Debug("Collected inventory baseline",
			"resource", resourceType,
			"resource_count", len(records),
			"raw_resource_count", len(execution.Resources),
			"id_collision_count", collisionCount,
			"resources_path", execution.ResourcesPath,
			"exit_code", execution.ExitCode,
			"had_error", baseline.Err != nil,
			"error", execution.Error,
		)
	}
	return baselines
}

func (p *CloudCustodianPlugin) buildResourcePayloadsForCheck(
	check CustodianCheck,
	execution CustodianExecutionResult,
	baseline *InventoryBaseline,
) []*StandardizedResourcePayload {
	matchedRecords := make([]ResourceRecord, 0, len(execution.Resources))
	for _, resource := range execution.Resources {
		matchedRecords = append(matchedRecords, p.buildResourceRecord(check.Resource, resource))
	}
	collisionIDs := mergeCollisionIDs(
		resourceIDCollisions(baseline.Records),
		resourceIDCollisions(matchedRecords),
	)
	baselineResources, baselineCollisionCount := disambiguateResourceRecords(baseline.Records, collisionIDs)
	matched, collisionCount := disambiguateResourceRecords(matchedRecords, collisionIDs)
	if baselineCollisionCount > 0 {
		p.Logger.Warn("Detected duplicate baseline resource identifiers; disambiguating with stable secondary keys",
			"check_name", check.Name,
			"resource", check.Resource,
			"id_collision_count", baselineCollisionCount,
		)
	}
	if collisionCount > 0 {
		p.Logger.Warn("Detected duplicate matched resource identifiers; disambiguating with stable secondary keys",
			"check_name", check.Name,
			"resource", check.Resource,
			"id_collision_count", collisionCount,
		)
	}

	resourceIDs := make([]string, 0, len(baselineResources)+len(matched))
	seen := map[string]bool{}
	for resourceID := range baselineResources {
		resourceIDs = append(resourceIDs, resourceID)
		seen[resourceID] = true
	}
	for resourceID := range matched {
		if seen[resourceID] {
			continue
		}
		resourceIDs = append(resourceIDs, resourceID)
	}
	slices.Sort(resourceIDs)

	payloads := make([]*StandardizedResourcePayload, 0, len(resourceIDs))
	for _, resourceID := range resourceIDs {
		record, existsInBaseline := baselineResources[resourceID]
		matchedRecord, isMatched := matched[resourceID]
		inventoryStatus := "baseline"
		if !existsInBaseline {
			record = matchedRecord
			inventoryStatus = "missing_from_baseline"
		}

		status := "compliant"
		if isMatched {
			status = "non_compliant"
		}

		payloads = append(payloads, buildResourcePayload(check, execution, record, StandardizedAssessment{
			Status:               status,
			Matched:              isMatched,
			InventoryStatus:      inventoryStatus,
			MatchedResourceCount: len(execution.Resources),
			ArtifactPath:         execution.ArtifactPath,
			ResourcesPath:        execution.ResourcesPath,
		}))
	}
	return payloads
}

type PayloadAssessmentStats struct {
	Compliant           int
	NonCompliant        int
	MissingFromBaseline int
}

func summarizePayloadAssessments(payloads []*StandardizedResourcePayload) PayloadAssessmentStats {
	stats := PayloadAssessmentStats{}
	for _, payload := range payloads {
		if payload == nil {
			continue
		}
		switch payload.Assessment.Status {
		case "compliant":
			stats.Compliant++
		case "non_compliant":
			stats.NonCompliant++
		}
		if payload.Assessment.InventoryStatus == "missing_from_baseline" {
			stats.MissingFromBaseline++
		}
	}
	return stats
}

func (p *CloudCustodianPlugin) evaluateResourcePolicies(
	ctx context.Context,
	payload *StandardizedResourcePayload,
	policyPaths []string,
) ([]*proto.Evidence, error, int) {
	p.Logger.Debug("Evaluating policy paths for resource",
		"check_name", payload.Check.Name,
		"resource_id", payload.Resource.ID,
		"assessment_status", payload.Assessment.Status,
		"policy_paths_count", len(policyPaths),
	)
	p.logPolicyPayload(payload)
	labels := resourcePolicyLabels(p.parsedConfig.PolicyLabels)
	labels["source"] = sourceCloudCustodian
	labels["tool"] = sourceCloudCustodian
	if _, exists := labels["provider"]; !exists {
		labels["provider"] = payload.Resource.Provider
	}
	labels["type"] = "resource"
	labels["check_name"] = payload.Check.Name
	labels["check_resource"] = payload.Check.Resource
	labels["check_status"] = payload.Execution.Status
	labels["resource_type"] = payload.Resource.Type
	labels["resource_id"] = payload.Resource.ID
	if payload.Resource.AccountID != "" {
		labels["account_id"] = payload.Resource.AccountID
	}
	if payload.Resource.Region != "" {
		labels["region"] = payload.Resource.Region
	}

	checkID := fmt.Sprintf("cloud-custodian-check/%s-%d", sanitizeIdentifier(payload.Check.Name), payload.Check.Index+1)
	providerID := fmt.Sprintf("cloud-provider/%s", sanitizeIdentifier(payload.Check.Provider))
	resourceSubjectID := fmt.Sprintf(
		"cloud-custodian-resource/%s/%s",
		url.PathEscape(payload.Resource.Type),
		url.PathEscape(payload.Resource.ID),
	)

	actors := []*proto.OriginActor{
		{
			Title: "The Continuous Compliance Framework",
			Type:  "assessment-platform",
			Links: []*proto.Link{
				{
					Href: "https://compliance-framework.github.io/docs/",
					Rel:  policyManager.Pointer("reference"),
					Text: policyManager.Pointer("The Continuous Compliance Framework"),
				},
			},
		},
		{
			Title: "Continuous Compliance Framework - Cloud Custodian Plugin",
			Type:  "tool",
			Links: []*proto.Link{
				{
					Href: "https://github.com/compliance-framework/plugin-cloud-custodian",
					Rel:  policyManager.Pointer("reference"),
					Text: policyManager.Pointer("The Continuous Compliance Framework Cloud Custodian Plugin"),
				},
			},
		},
	}

	components := []*proto.Component{
		{
			Identifier:  "cloud-custodian/runtime",
			Type:        "tool",
			Title:       "Cloud Custodian Runtime",
			Description: "Cloud Custodian CLI runtime used to execute provider policies in dry-run mode.",
			Purpose:     "To execute policy checks without mutating cloud resources.",
		},
		{
			Identifier:  providerID,
			Type:        "service",
			Title:       fmt.Sprintf("Cloud Provider: %s", payload.Check.Provider),
			Description: "Cloud service endpoint evaluated by the Cloud Custodian check.",
			Purpose:     "To provide resource inventory and configuration data for compliance checks.",
		},
	}

	inventory := []*proto.InventoryItem{
		{
			Identifier: checkID,
			Type:       "cloud-custodian-check",
			Title:      fmt.Sprintf("Cloud Custodian Check %s", payload.Check.Name),
			// Local temp artifact paths are intentionally not exposed in evidence links
			// because they are ephemeral and not portable for downstream consumers.
			ImplementedComponents: []*proto.InventoryItemImplementedComponent{
				{Identifier: "cloud-custodian/runtime"},
				{Identifier: providerID},
			},
		},
	}

	subjects := []*proto.Subject{
		{
			Type:       proto.SubjectType_SUBJECT_TYPE_RESOURCE,
			Identifier: resourceSubjectID,
			Links: []*proto.Link{
				{
					Href: payload.Resource.ID,
					Rel:  policyManager.Pointer("related"),
					Text: policyManager.Pointer("Cloud resource identifier"),
				},
			},
			Props: []*proto.Property{
				{Name: "resource_id", Value: payload.Resource.ID},
			},
		},
		{Type: proto.SubjectType_SUBJECT_TYPE_INVENTORY_ITEM, Identifier: checkID},
		{Type: proto.SubjectType_SUBJECT_TYPE_COMPONENT, Identifier: providerID},
	}

	activities := []*proto.Activity{
		{
			Title: "Execute Cloud Custodian Check",
			Steps: []*proto.Step{
				{Title: "Load Policy", Description: "Load one Cloud Custodian policy entry from the configured policy document."},
				{Title: "Run Dry-Run Check", Description: "Execute Cloud Custodian using --dryrun and capture generated artifacts."},
				{Title: "Build Resource Payload", Description: "Compare policy matches with inventory baseline and build one standardized OPA input for this resource."},
			},
		},
		{
			Title: "Evaluate OPA Policy Bundles",
			Steps: []*proto.Step{
				{Title: "Evaluate Resource Payload", Description: "Run policy bundles against the standardized Cloud Custodian resource payload."},
			},
		},
	}

	allEvidences := make([]*proto.Evidence, 0)
	var accumulatedErrors error
	successfulRuns := 0

	for _, policyPath := range policyPaths {
		p.Logger.Trace("Running policy path for resource", "check_name", payload.Check.Name, "resource_id", payload.Resource.ID, "policy_path", policyPath)
		evidences, err := p.evaluator.Generate(
			ctx,
			policyPath,
			labels,
			subjects,
			components,
			inventory,
			actors,
			activities,
			payload,
		)
		addResourceExplorerLinkToEvidence(evidences, payload, p.Logger)
		allEvidences = append(allEvidences, evidences...)
		if err != nil {
			p.Logger.Warn("Policy path evaluation failed for resource",
				"check_name", payload.Check.Name,
				"resource_id", payload.Resource.ID,
				"policy_path", policyPath,
				"error", err,
			)
			accumulatedErrors = errors.Join(accumulatedErrors, fmt.Errorf("policy %s failed for check %s resource %s: %w", policyPath, payload.Check.Name, payload.Resource.ID, err))
			continue
		}
		p.Logger.Trace("Policy path evaluation succeeded for resource",
			"check_name", payload.Check.Name,
			"resource_id", payload.Resource.ID,
			"policy_path", policyPath,
			"evidence_count", len(evidences),
		)
		successfulRuns++
	}

	p.Logger.Debug("Completed policy path loop for resource",
		"check_name", payload.Check.Name,
		"resource_id", payload.Resource.ID,
		"successful_runs", successfulRuns,
		"evidence_count", len(allEvidences),
		"had_errors", accumulatedErrors != nil,
	)
	return allEvidences, accumulatedErrors, successfulRuns
}

func addResourceExplorerLinkToEvidence(evidences []*proto.Evidence, payload *StandardizedResourcePayload, logger hclog.Logger) {
	link, reason := awsResourceExplorerLink(payload)
	if link == nil {
		logResourceExplorerLinkSkipped(logger, payload, reason)
		return
	}
	for _, evidence := range evidences {
		if evidence == nil {
			continue
		}
		evidence.Links = append(evidence.Links, &proto.Link{
			Href: link.Href,
			Rel:  link.Rel,
			Text: link.Text,
		})
	}
}

func logResourceExplorerLinkSkipped(logger hclog.Logger, payload *StandardizedResourcePayload, reason string) {
	if logger == nil {
		return
	}
	if reason == "" {
		reason = "unknown reason"
	}
	var checkName, resourceID, resourceType, provider string
	if payload != nil {
		checkName = payload.Check.Name
		resourceID = payload.Resource.ID
		resourceType = payload.Resource.Type
		provider = payload.Resource.Provider
	}
	message := "Skipping AWS Resource Explorer evidence link generation"
	args := []interface{}{
		"reason", reason,
		"check_name", checkName,
		"resource_id", resourceID,
		"resource_type", resourceType,
		"provider", provider,
	}
	if shouldWarnResourceExplorerLinkSkipped(reason) {
		logger.Warn(message, args...)
		return
	}
	logger.Debug(message, args...)
}

func shouldWarnResourceExplorerLinkSkipped(reason string) bool {
	switch reason {
	case "resource id is not an ARN", "provider is not aws":
		return false
	default:
		return true
	}
}

func awsResourceExplorerLink(payload *StandardizedResourcePayload) (*proto.Link, string) {
	href, reason := awsResourceExplorerURLWithReason(payload)
	if href != "" {
		return &proto.Link{
			Href: href,
			Rel:  policyManager.Pointer("reference"),
			Text: policyManager.Pointer("Open in AWS Resource Explorer"),
		}, ""
	}
	if reason == "" {
		reason = "unknown reason"
	}
	return nil, reason
}

func awsResourceExplorerURL(payload *StandardizedResourcePayload) string {
	href, _ := awsResourceExplorerURLWithReason(payload)
	return href
}

func awsResourceExplorerURLWithReason(payload *StandardizedResourcePayload) (string, string) {
	resourceARN, reason := awsResourceExplorerResourceARN(payload)
	if resourceARN == "" {
		return "", reason
	}

	partition := awsPartitionFromARN(resourceARN)
	region := awsRegionFromARN(resourceARN)
	if region == "" && payload.Resource.Region != "" && payload.Resource.Region != "global" && awsPartitionForRegion(payload.Resource.Region) == partition {
		region = payload.Resource.Region
	}
	if region == "" {
		region = awsDefaultConsoleRegionForPartition(partition)
	}

	query := "id:" + resourceARN
	consoleDomain := awsConsoleDomainForPartition(partition)
	return "https://" + consoleDomain + "/resource-explorer/home?region=" + url.QueryEscape(region) + "#/search?query=" + url.QueryEscape(query), ""
}

func awsResourceExplorerResourceARN(payload *StandardizedResourcePayload) (string, string) {
	if payload == nil {
		return "", "payload is nil"
	}
	if !strings.EqualFold(payload.Resource.Provider, "aws") {
		return "", "provider is not aws"
	}

	resourceID := strings.TrimSpace(payload.Resource.ID)
	if resourceID == "" {
		return "", "resource id is empty"
	}
	if strings.HasPrefix(resourceID, "arn:") {
		return resourceID, ""
	}
	if strings.EqualFold(payload.Resource.Type, "aws.s3") {
		bucketName, ok := normalizeS3BucketName(resourceID)
		if !ok {
			return "", "aws.s3 resource id is not a valid S3 bucket name"
		}
		return "arn:" + awsPartitionForResource(payload.Resource) + ":s3:::" + bucketName, ""
	}
	return "", "resource id is not an ARN"
}

func awsPartitionForResource(resource StandardizedResourceInfo) string {
	if partition := awsPartitionFromARN(resource.AccountID); partition != "" {
		return partition
	}
	return awsPartitionForRegion(resource.Region)
}

func awsPartitionFromARN(arnValue string) string {
	parts := strings.SplitN(strings.TrimSpace(arnValue), ":", 3)
	if len(parts) >= 2 && parts[0] == "arn" && strings.TrimSpace(parts[1]) != "" {
		return strings.TrimSpace(parts[1])
	}
	return ""
}

func awsConsoleDomainForPartition(partition string) string {
	switch partition {
	case "aws-cn":
		return "console.amazonaws.cn"
	case "aws-us-gov":
		return "console.amazonaws-us-gov.com"
	default:
		return "console.aws.amazon.com"
	}
}

func awsDefaultConsoleRegionForPartition(partition string) string {
	switch partition {
	case "aws-cn":
		return "cn-north-1"
	case "aws-us-gov":
		return "us-gov-west-1"
	default:
		return "us-east-1"
	}
}

func awsPartitionForRegion(region string) string {
	region = strings.TrimSpace(strings.ToLower(region))
	switch {
	case strings.HasPrefix(region, "cn-"):
		return "aws-cn"
	case strings.HasPrefix(region, "us-gov-"):
		return "aws-us-gov"
	default:
		return "aws"
	}
}

func normalizeS3BucketName(resourceID string) (string, bool) {
	bucketName := strings.TrimSpace(resourceID)
	bucketName = strings.TrimPrefix(bucketName, "s3://")
	if slash := strings.Index(bucketName, "/"); slash >= 0 {
		bucketName = bucketName[:slash]
	}
	if !isValidS3BucketName(bucketName) {
		return "", false
	}
	return bucketName, true
}

func isValidS3BucketName(bucketName string) bool {
	if len(bucketName) < 3 || len(bucketName) > 63 {
		return false
	}
	if net.ParseIP(bucketName) != nil {
		return false
	}
	if strings.Contains(bucketName, "..") || strings.Contains(bucketName, ".-") || strings.Contains(bucketName, "-.") {
		return false
	}
	for index, char := range bucketName {
		valid := char >= 'a' && char <= 'z' || char >= '0' && char <= '9' || char == '.' || char == '-'
		if !valid {
			return false
		}
		if index == 0 || index == len(bucketName)-1 {
			if !(char >= 'a' && char <= 'z' || char >= '0' && char <= '9') {
				return false
			}
		}
	}
	return true
}

func awsRegionFromARN(arn string) string {
	parts := strings.SplitN(arn, ":", 6)
	if len(parts) < 6 {
		return ""
	}
	return strings.TrimSpace(parts[3])
}

func resourcePolicyLabels(policyLabels map[string]string) map[string]string {
	labels := map[string]string{}
	for key, value := range policyLabels {
		if isReservedResourceLabel(key) {
			continue
		}
		labels[key] = value
	}
	return labels
}

func isReservedResourceLabel(label string) bool {
	switch label {
	case "assessment", "assessment_status", "check_provider":
		return true
	default:
		return false
	}
}

func formatExecutionFailure(checkName string, execution CustodianExecutionResult) error {
	switch {
	case execution.Error != "" && execution.Err != nil:
		return fmt.Errorf("custodian policy execution failed for check %s: %s: %w", checkName, execution.Error, execution.Err)
	case execution.Error != "":
		return fmt.Errorf("custodian policy execution failed for check %s: %s", checkName, execution.Error)
	case execution.Err != nil:
		return fmt.Errorf("custodian policy execution failed for check %s: %w", checkName, execution.Err)
	default:
		return fmt.Errorf("custodian policy execution failed for check %s", checkName)
	}
}

func formatExecutionDiagnosticWarnings(messages []string) error {
	var err error
	for _, message := range messages {
		message = strings.TrimSpace(message)
		if message == "" {
			continue
		}
		err = errors.Join(err, fmt.Errorf("custodian policy execution warning: %s", message))
	}
	return err
}

func (p *CloudCustodianPlugin) logPolicyPayload(payload *StandardizedResourcePayload) {
	if payload == nil || !p.Logger.IsDebug() {
		return
	}

	p.Logger.Debug("Policy payload",
		"check_name", payload.Check.Name,
		"resource_id", payload.Resource.ID,
		"assessment_status", payload.Assessment.Status,
		"resource_type", payload.Resource.Type,
		"provider", payload.Resource.Provider,
		"debug_dump_payloads", p.parsedConfig != nil && p.parsedConfig.DebugDumpPayloads,
	)
}

func (p *CloudCustodianPlugin) dumpStandardizedPayload(payload *StandardizedResourcePayload) error {
	if payload == nil {
		return errors.New("payload is nil")
	}
	if p.parsedConfig == nil || !p.parsedConfig.DebugDumpPayloads {
		return nil
	}

	content, err := json.MarshalIndent(payload, "", "  ")
	if err != nil {
		return fmt.Errorf("marshal payload: %w", err)
	}

	sanitizedCheckName := sanitizeIdentifier(payload.Check.Name)
	sanitizedResourceID := sanitizeIdentifier(payload.Resource.ID)
	if len(sanitizedResourceID) > 50 {
		shortHash := hashResource(payload.Resource.ID)[:12]
		sanitizedResourceID = sanitizedResourceID[:50] + "-" + shortHash
	}
	fileName := fmt.Sprintf("%03d-%s-%s-%d.json",
		payload.Check.Index+1,
		sanitizedCheckName,
		sanitizedResourceID,
		time.Now().UTC().UnixNano(),
	)
	outputPath := filepath.Join(p.parsedConfig.DebugPayloadOutputDir, fileName)
	if err := os.WriteFile(outputPath, content, 0o600); err != nil {
		return fmt.Errorf("write payload file %s: %w", outputPath, err)
	}

	p.Logger.Debug("Wrote standardized payload debug file",
		"check_name", payload.Check.Name,
		"resource_id", payload.Resource.ID,
		"output_path", outputPath,
		"bytes", len(content),
	)
	return nil
}

func newCheckErrorExecution(messages []string) CustodianExecutionResult {
	now := time.Now().UTC()
	normalized := make([]string, 0, len(messages))
	for _, message := range messages {
		if strings.TrimSpace(message) == "" {
			continue
		}
		normalized = append(normalized, message)
	}
	joined := strings.Join(normalized, "; ")
	if joined == "" {
		joined = "check failed"
		normalized = []string{joined}
	}
	err := errors.New(joined)
	return CustodianExecutionResult{
		StartedAt: now,
		EndedAt:   now,
		ExitCode:  -1,
		Error:     joined,
		Errors:    normalized,
		Err:       err,
		Resources: []interface{}{},
	}
}

func sanitizeIdentifier(in string) string {
	trimmed := strings.TrimSpace(strings.ToLower(in))
	if trimmed == "" {
		return "unknown"
	}

	builder := strings.Builder{}
	prevDash := false
	for _, r := range trimmed {
		isAlphaNum := (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9')
		if isAlphaNum {
			builder.WriteRune(r)
			prevDash = false
			continue
		}
		if !prevDash {
			builder.WriteRune('-')
			prevDash = true
		}
	}

	out := strings.Trim(builder.String(), "-")
	if out == "" {
		return "unknown"
	}
	return out
}

func sortedKeys(input map[string]string) []string {
	keys := make([]string, 0, len(input))
	for k := range input {
		keys = append(keys, k)
	}
	slices.Sort(keys)
	return keys
}

func main() {
	logger := hclog.New(&hclog.LoggerOptions{
		Level:      hclog.Trace,
		JSONFormat: true,
	})

	plugin := &CloudCustodianPlugin{Logger: logger}

	logger.Info("Starting Cloud Custodian Plugin")
	goplugin.Serve(&goplugin.ServeConfig{
		HandshakeConfig: runner.HandshakeConfig,
		Plugins: map[string]goplugin.Plugin{
			"runner": &runner.RunnerV2GRPCPlugin{Impl: plugin},
		},
		GRPCServer: goplugin.DefaultGRPCServer,
	})
}