426 lines
12 KiB
Python
426 lines
12 KiB
Python
"""Core data models for the IaC Reverse Engineering tool.
|
|
|
|
Contains enums, dataclasses, and type definitions used across all components
|
|
of the pipeline: Scanner, Dependency Resolver, Code Generator, State Builder,
|
|
Validator, and Incremental Scan Engine.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Optional
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Enums
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ProviderType(Enum):
|
|
"""Supported on-premises infrastructure provider types."""
|
|
|
|
DOCKER_SWARM = "docker_swarm"
|
|
KUBERNETES = "kubernetes"
|
|
SYNOLOGY = "synology"
|
|
HARVESTER = "harvester"
|
|
BARE_METAL = "bare_metal"
|
|
WINDOWS = "windows"
|
|
|
|
|
|
class PlatformCategory(Enum):
|
|
"""Categorizes providers by their infrastructure model."""
|
|
|
|
CONTAINER_ORCHESTRATION = "container" # Docker Swarm, Kubernetes
|
|
STORAGE_APPLIANCE = "storage" # Synology Disk Station
|
|
HCI = "hci" # SUSE Harvester (Hyper-Converged Infrastructure)
|
|
BARE_METAL = "bare_metal" # Physical servers (Linux)
|
|
WINDOWS = "windows" # Standalone Windows machines
|
|
|
|
|
|
PROVIDER_PLATFORM_MAP: dict[ProviderType, PlatformCategory] = {
|
|
ProviderType.DOCKER_SWARM: PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
ProviderType.KUBERNETES: PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
ProviderType.SYNOLOGY: PlatformCategory.STORAGE_APPLIANCE,
|
|
ProviderType.HARVESTER: PlatformCategory.HCI,
|
|
ProviderType.BARE_METAL: PlatformCategory.BARE_METAL,
|
|
ProviderType.WINDOWS: PlatformCategory.WINDOWS,
|
|
}
|
|
|
|
|
|
class CpuArchitecture(Enum):
|
|
"""CPU architecture of the host or resource."""
|
|
|
|
AMD64 = "amd64"
|
|
ARM = "arm"
|
|
AARCH64 = "aarch64"
|
|
|
|
|
|
class ChangeType(Enum):
|
|
"""Classification of resource changes between scan runs."""
|
|
|
|
ADDED = "added"
|
|
REMOVED = "removed"
|
|
MODIFIED = "modified"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provider supported resource types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PROVIDER_SUPPORTED_RESOURCE_TYPES: dict[ProviderType, list[str]] = {
|
|
ProviderType.DOCKER_SWARM: [
|
|
"docker_service",
|
|
"docker_network",
|
|
"docker_volume",
|
|
"docker_config",
|
|
"docker_secret",
|
|
],
|
|
ProviderType.KUBERNETES: [
|
|
"kubernetes_deployment",
|
|
"kubernetes_service",
|
|
"kubernetes_ingress",
|
|
"kubernetes_config_map",
|
|
"kubernetes_persistent_volume",
|
|
"kubernetes_namespace",
|
|
],
|
|
ProviderType.SYNOLOGY: [
|
|
"synology_shared_folder",
|
|
"synology_volume",
|
|
"synology_storage_pool",
|
|
"synology_replication_task",
|
|
"synology_user",
|
|
],
|
|
ProviderType.HARVESTER: [
|
|
"harvester_virtualmachine",
|
|
"harvester_volume",
|
|
"harvester_image",
|
|
"harvester_network",
|
|
],
|
|
ProviderType.BARE_METAL: [
|
|
"bare_metal_hardware",
|
|
"bare_metal_bmc_config",
|
|
"bare_metal_network_interface",
|
|
"bare_metal_raid_config",
|
|
],
|
|
ProviderType.WINDOWS: [
|
|
"windows_service",
|
|
"windows_scheduled_task",
|
|
"windows_iis_site",
|
|
"windows_iis_app_pool",
|
|
"windows_network_adapter",
|
|
"windows_firewall_rule",
|
|
"windows_installed_software",
|
|
"windows_feature",
|
|
"windows_hyperv_vm",
|
|
"windows_hyperv_switch",
|
|
"windows_dns_record",
|
|
"windows_local_user",
|
|
"windows_local_group",
|
|
],
|
|
}
|
|
|
|
MAX_RESOURCE_TYPE_FILTERS = 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scanner dataclasses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class ScanProfile:
|
|
"""Configuration for a single infrastructure scan."""
|
|
|
|
provider: ProviderType
|
|
credentials: dict[str, str]
|
|
endpoints: Optional[list[str]] = None
|
|
resource_type_filters: Optional[list[str]] = None
|
|
authentik_token: Optional[str] = None
|
|
|
|
def validate(self) -> list[str]:
|
|
"""Returns list of validation errors, empty if valid.
|
|
|
|
Validates:
|
|
- credentials must not be empty
|
|
- resource_type_filters must have at most MAX_RESOURCE_TYPE_FILTERS entries
|
|
- resource_type_filters entries must be supported by the provider
|
|
"""
|
|
errors: list[str] = []
|
|
|
|
if not self.credentials:
|
|
errors.append("credentials must not be empty")
|
|
|
|
if self.resource_type_filters is not None:
|
|
if len(self.resource_type_filters) > MAX_RESOURCE_TYPE_FILTERS:
|
|
errors.append(
|
|
f"resource_type_filters must have at most "
|
|
f"{MAX_RESOURCE_TYPE_FILTERS} entries, "
|
|
f"got {len(self.resource_type_filters)}"
|
|
)
|
|
|
|
supported = set(PROVIDER_SUPPORTED_RESOURCE_TYPES[self.provider])
|
|
unsupported = [
|
|
rt for rt in self.resource_type_filters if rt not in supported
|
|
]
|
|
if unsupported:
|
|
errors.append(
|
|
f"unsupported resource types for provider "
|
|
f"'{self.provider.value}': {unsupported}"
|
|
)
|
|
|
|
return errors
|
|
|
|
@property
|
|
def platform_category(self) -> PlatformCategory:
|
|
"""Return the platform category for this profile's provider."""
|
|
return PROVIDER_PLATFORM_MAP[self.provider]
|
|
|
|
|
|
@dataclass
|
|
class DiscoveredResource:
|
|
"""A single resource discovered from an infrastructure provider."""
|
|
|
|
resource_type: str
|
|
unique_id: str
|
|
name: str
|
|
provider: ProviderType
|
|
platform_category: PlatformCategory
|
|
architecture: CpuArchitecture
|
|
endpoint: str
|
|
attributes: dict
|
|
raw_references: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class ScanResult:
|
|
"""Complete result of a scan operation."""
|
|
|
|
resources: list[DiscoveredResource]
|
|
warnings: list[str]
|
|
errors: list[str]
|
|
scan_timestamp: str
|
|
profile_hash: str
|
|
is_partial: bool = False
|
|
|
|
|
|
@dataclass
|
|
class ScanProgress:
|
|
"""Progress update during a scan operation."""
|
|
|
|
current_resource_type: str
|
|
resources_discovered: int
|
|
resource_types_completed: int
|
|
total_resource_types: int
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dependency Resolver dataclasses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class ResourceRelationship:
|
|
"""A relationship between two discovered resources."""
|
|
|
|
source_id: str
|
|
target_id: str
|
|
relationship_type: str # "parent-child", "reference", "dependency"
|
|
source_attribute: str
|
|
|
|
|
|
@dataclass
|
|
class UnresolvedReference:
|
|
"""A reference that could not be resolved to a known resource."""
|
|
|
|
source_resource_id: str
|
|
source_attribute: str
|
|
referenced_id: str
|
|
suggested_resolution: str # "data_source" or "variable"
|
|
|
|
|
|
@dataclass
|
|
class CycleReport:
|
|
"""Report of a detected circular dependency with resolution suggestions."""
|
|
|
|
cycle: list[str] # Resource IDs forming the cycle
|
|
suggested_break: tuple[str, str] # (source_id, target_id) edge to break
|
|
break_relationship_type: str # Type of the relationship to break
|
|
resolution_strategy: str # Human-readable suggestion for resolution
|
|
|
|
|
|
@dataclass
|
|
class DependencyGraph:
|
|
"""Complete dependency graph of discovered resources."""
|
|
|
|
resources: list[DiscoveredResource]
|
|
relationships: list[ResourceRelationship]
|
|
topological_order: list[str]
|
|
cycles: list[list[str]]
|
|
unresolved_references: list[UnresolvedReference]
|
|
cycle_reports: list[CycleReport] = field(default_factory=list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Code Generator dataclasses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class GeneratedFile:
|
|
"""A single generated Terraform HCL file."""
|
|
|
|
filename: str
|
|
content: str
|
|
resource_count: int
|
|
|
|
|
|
@dataclass
|
|
class ExtractedVariable:
|
|
"""A Terraform variable extracted from common resource values."""
|
|
|
|
name: str
|
|
type_expr: str
|
|
default_value: str
|
|
description: str
|
|
used_by: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class CodeGenerationResult:
|
|
"""Complete result of code generation."""
|
|
|
|
resource_files: list[GeneratedFile]
|
|
variables_file: GeneratedFile
|
|
provider_file: GeneratedFile
|
|
outputs_file: Optional[GeneratedFile] = None
|
|
skipped_resources: list[tuple[str, str]] = field(default_factory=list)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# State Builder dataclasses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class StateEntry:
|
|
"""A single resource entry in the Terraform state file."""
|
|
|
|
resource_type: str
|
|
resource_name: str
|
|
provider_id: str
|
|
attributes: dict
|
|
sensitive_attributes: list[str] = field(default_factory=list)
|
|
schema_version: int = 0
|
|
dependencies: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class StateFile:
|
|
"""Terraform state file representation (format version 4)."""
|
|
|
|
version: int = 4
|
|
terraform_version: str = ""
|
|
serial: int = 1
|
|
lineage: str = ""
|
|
resources: list[StateEntry] = field(default_factory=list)
|
|
|
|
def to_json(self) -> str:
|
|
"""Serialize to Terraform state JSON format."""
|
|
import json
|
|
import uuid
|
|
|
|
lineage = self.lineage or str(uuid.uuid4())
|
|
|
|
state_resources = []
|
|
for entry in self.resources:
|
|
state_resources.append(
|
|
{
|
|
"mode": "managed",
|
|
"type": entry.resource_type,
|
|
"name": entry.resource_name,
|
|
"provider": f'provider["registry.terraform.io/hashicorp/{entry.resource_type.split("_")[0]}"]',
|
|
"instances": [
|
|
{
|
|
"schema_version": entry.schema_version,
|
|
"attributes": {
|
|
"id": entry.provider_id,
|
|
**entry.attributes,
|
|
},
|
|
"sensitive_attributes": entry.sensitive_attributes,
|
|
"dependencies": entry.dependencies,
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
state = {
|
|
"version": self.version,
|
|
"terraform_version": self.terraform_version,
|
|
"serial": self.serial,
|
|
"lineage": lineage,
|
|
"outputs": {},
|
|
"resources": state_resources,
|
|
}
|
|
|
|
return json.dumps(state, indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validator dataclasses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class PlannedChange:
|
|
"""A single planned change reported by terraform plan."""
|
|
|
|
resource_address: str
|
|
change_type: str # "add", "modify", "destroy"
|
|
details: str
|
|
|
|
|
|
@dataclass
|
|
class ValidationError:
|
|
"""A validation error from terraform validate or plan."""
|
|
|
|
file: str
|
|
message: str
|
|
line: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class ValidationResult:
|
|
"""Complete result of terraform validation."""
|
|
|
|
init_success: bool
|
|
validate_success: bool
|
|
plan_success: bool
|
|
planned_changes: list[PlannedChange] = field(default_factory=list)
|
|
errors: list[ValidationError] = field(default_factory=list)
|
|
correction_attempts: int = 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Incremental Scan dataclasses
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class ResourceChange:
|
|
"""A single resource change between scan runs."""
|
|
|
|
resource_id: str
|
|
resource_type: str
|
|
resource_name: str
|
|
change_type: ChangeType
|
|
changed_attributes: Optional[dict] = None
|
|
|
|
|
|
@dataclass
|
|
class ChangeSummary:
|
|
"""Summary of changes between two scan runs."""
|
|
|
|
added_count: int
|
|
removed_count: int
|
|
modified_count: int
|
|
changes: list[ResourceChange] = field(default_factory=list)
|