|
1
|
""" |
|
2
|
Terraform enricher for cross-file module resolution and resource linking. |
|
3
|
|
|
4
|
Promotes and links Terraform graph nodes: |
|
5
|
- Cross-file variable references (REFERENCES edges) |
|
6
|
- Module source resolution (DEPENDS_ON edges to local source dirs) |
|
7
|
- Provider grouping (BELONGS_TO edges to provider nodes) |
|
8
|
""" |
|
9
|
|
|
10
|
from navegador.enrichment.base import EnrichmentResult, FrameworkEnricher |
|
11
|
from navegador.graph.store import GraphStore |
|
12
|
|
|
13
|
# Common Terraform provider prefixes and their canonical provider names |
|
14
|
_PROVIDER_PREFIXES = { |
|
15
|
"aws_": "aws", |
|
16
|
"google_": "google", |
|
17
|
"azurerm_": "azurerm", |
|
18
|
"azuread_": "azuread", |
|
19
|
"kubernetes_": "kubernetes", |
|
20
|
"helm_": "helm", |
|
21
|
"vault_": "vault", |
|
22
|
"datadog_": "datadog", |
|
23
|
"cloudflare_": "cloudflare", |
|
24
|
"digitalocean_": "digitalocean", |
|
25
|
"github_": "github", |
|
26
|
"null_": "null", |
|
27
|
"random_": "random", |
|
28
|
"local_": "local", |
|
29
|
"tls_": "tls", |
|
30
|
"archive_": "archive", |
|
31
|
"external_": "external", |
|
32
|
"template_": "template", |
|
33
|
"time_": "time", |
|
34
|
} |
|
35
|
|
|
36
|
|
|
37
|
class TerraformEnricher(FrameworkEnricher): |
|
38
|
"""Enriches a navegador graph with Terraform-specific semantics.""" |
|
39
|
|
|
40
|
def __init__(self, store: GraphStore) -> None: |
|
41
|
super().__init__(store) |
|
42
|
|
|
43
|
# ── Identity ────────────────────────────────────────────────────────────── |
|
44
|
|
|
45
|
@property |
|
46
|
def framework_name(self) -> str: |
|
47
|
return "terraform" |
|
48
|
|
|
49
|
@property |
|
50
|
def detection_patterns(self) -> list[str]: |
|
51
|
return [] # No import nodes for Terraform |
|
52
|
|
|
53
|
@property |
|
54
|
def detection_files(self) -> list[str]: |
|
55
|
return ["main.tf", "variables.tf", "outputs.tf", "providers.tf"] |
|
56
|
|
|
57
|
# ── Enrichment ──────────────────────────────────────────────────────────── |
|
58
|
|
|
59
|
def enrich(self) -> EnrichmentResult: |
|
60
|
result = EnrichmentResult() |
|
61
|
|
|
62
|
var_refs = self._enrich_variable_references() |
|
63
|
result.edges_added += var_refs |
|
64
|
result.patterns_found["variable_references"] = var_refs |
|
65
|
|
|
66
|
module_deps = self._enrich_module_sources() |
|
67
|
result.edges_added += module_deps |
|
68
|
result.patterns_found["module_sources"] = module_deps |
|
69
|
|
|
70
|
provider_links = self._enrich_provider_grouping() |
|
71
|
result.edges_added += provider_links |
|
72
|
result.patterns_found["provider_grouping"] = provider_links |
|
73
|
|
|
74
|
return result |
|
75
|
|
|
76
|
# ── Pattern helpers ─────────────────────────────────────────────────────── |
|
77
|
|
|
78
|
def _enrich_variable_references(self) -> int: |
|
79
|
"""Find terraform_variable and terraform_output nodes that reference |
|
80
|
variables defined in other files, and create REFERENCES edges.""" |
|
81
|
edges_added = 0 |
|
82
|
|
|
83
|
# Find all terraform_variable nodes |
|
84
|
var_result = self.store.query( |
|
85
|
"MATCH (v) WHERE v.semantic_type = $var_type RETURN v.name, v.file_path", |
|
86
|
{"var_type": "terraform_variable"}, |
|
87
|
) |
|
88
|
var_rows = var_result.result_set or [] |
|
89
|
var_by_name: dict[str, list[str]] = {} |
|
90
|
for row in var_rows: |
|
91
|
name, file_path = row[0], row[1] |
|
92
|
if name and file_path: |
|
93
|
var_by_name.setdefault(name, []).append(file_path) |
|
94
|
|
|
95
|
# Find terraform_output nodes and check if they reference variables |
|
96
|
# from other files (outputs often reference var.xxx) |
|
97
|
output_result = self.store.query( |
|
98
|
"MATCH (o) WHERE o.semantic_type = $out_type RETURN o.name, o.file_path", |
|
99
|
{"out_type": "terraform_output"}, |
|
100
|
) |
|
101
|
output_rows = output_result.result_set or [] |
|
102
|
for row in output_rows: |
|
103
|
out_name, out_file = row[0], row[1] |
|
104
|
if not (out_name and out_file): |
|
105
|
continue |
|
106
|
|
|
107
|
# Check CALLS or REFERENCES edges from this output to variables |
|
108
|
ref_result = self.store.query( |
|
109
|
"MATCH (o)-[:CALLS]->(target) " |
|
110
|
"WHERE o.name = $name AND o.file_path = $path " |
|
111
|
"RETURN target.name, target.file_path", |
|
112
|
{"name": out_name, "path": out_file}, |
|
113
|
) |
|
114
|
ref_rows = ref_result.result_set or [] |
|
115
|
for ref_row in ref_rows: |
|
116
|
target_name, target_path = ref_row[0], ref_row[1] |
|
117
|
if target_name and target_path and target_path != out_file: |
|
118
|
self._add_semantic_edge(out_name, "REFERENCES", target_name) |
|
119
|
edges_added += 1 |
|
120
|
|
|
121
|
# Also link variables in different files that share the same name |
|
122
|
# (e.g. variables.tf defines var, main.tf uses it) |
|
123
|
for var_name, paths in var_by_name.items(): |
|
124
|
if len(paths) <= 1: |
|
125
|
continue |
|
126
|
# Find nodes in other files that reference this variable |
|
127
|
for path in paths: |
|
128
|
ref_result = self.store.query( |
|
129
|
"MATCH (n) WHERE n.file_path <> $path " |
|
130
|
"AND n.name = $name " |
|
131
|
"AND n.semantic_type = $var_type " |
|
132
|
"RETURN n.name, n.file_path", |
|
133
|
{"path": path, "name": var_name, "var_type": "terraform_variable"}, |
|
134
|
) |
|
135
|
ref_rows = ref_result.result_set or [] |
|
136
|
for ref_row in ref_rows: |
|
137
|
ref_name = ref_row[0] |
|
138
|
if ref_name: |
|
139
|
self._add_semantic_edge(var_name, "REFERENCES", ref_name) |
|
140
|
edges_added += 1 |
|
141
|
|
|
142
|
return edges_added |
|
143
|
|
|
144
|
def _enrich_module_sources(self) -> int: |
|
145
|
"""Find terraform_module nodes with local source paths and create |
|
146
|
DEPENDS_ON edges to File nodes in the referenced directory.""" |
|
147
|
edges_added = 0 |
|
148
|
|
|
149
|
# Find Module nodes with terraform_module semantic type |
|
150
|
module_result = self.store.query( |
|
151
|
"MATCH (m) WHERE m.semantic_type = $mod_type RETURN m.name, m.file_path", |
|
152
|
{"mod_type": "terraform_module"}, |
|
153
|
) |
|
154
|
module_rows = module_result.result_set or [] |
|
155
|
|
|
156
|
for row in module_rows: |
|
157
|
mod_name, mod_file = row[0], row[1] |
|
158
|
if not (mod_name and mod_file): |
|
159
|
continue |
|
160
|
|
|
161
|
# Check for CALLS edges that may point to the source path, |
|
162
|
# or look for a source property on the node |
|
163
|
source_result = self.store.query( |
|
164
|
"MATCH (m)-[:CALLS]->(target) " |
|
165
|
"WHERE m.name = $name AND m.file_path = $path " |
|
166
|
"RETURN target.name, target.file_path", |
|
167
|
{"name": mod_name, "path": mod_file}, |
|
168
|
) |
|
169
|
source_rows = source_result.result_set or [] |
|
170
|
|
|
171
|
for source_row in source_rows: |
|
172
|
target_name, target_path = source_row[0], source_row[1] |
|
173
|
if target_name and target_path: |
|
174
|
self._add_semantic_edge(mod_name, "DEPENDS_ON", target_name) |
|
175
|
edges_added += 1 |
|
176
|
continue |
|
177
|
|
|
178
|
# Fallback: look for File nodes whose path contains the module name |
|
179
|
# (local modules are often in ./modules/<name>/) |
|
180
|
file_result = self.store.query( |
|
181
|
"MATCH (f:File) WHERE f.file_path CONTAINS $fragment RETURN f.name", |
|
182
|
{"fragment": f"/modules/{mod_name}/"}, |
|
183
|
) |
|
184
|
file_rows = file_result.result_set or [] |
|
185
|
for file_row in file_rows: |
|
186
|
target_name = file_row[0] |
|
187
|
if target_name: |
|
188
|
self._add_semantic_edge(mod_name, "DEPENDS_ON", target_name) |
|
189
|
edges_added += 1 |
|
190
|
|
|
191
|
return edges_added |
|
192
|
|
|
193
|
def _enrich_provider_grouping(self) -> int: |
|
194
|
"""Group Terraform resources by their provider prefix and create |
|
195
|
BELONGS_TO edges from resources to provider nodes.""" |
|
196
|
edges_added = 0 |
|
197
|
|
|
198
|
# Find all terraform_resource nodes |
|
199
|
resource_result = self.store.query( |
|
200
|
"MATCH (r) WHERE r.semantic_type = $res_type RETURN r.name, r.file_path", |
|
201
|
{"res_type": "terraform_resource"}, |
|
202
|
) |
|
203
|
resource_rows = resource_result.result_set or [] |
|
204
|
|
|
205
|
for row in resource_rows: |
|
206
|
res_name, res_file = row[0], row[1] |
|
207
|
if not (res_name and res_file): |
|
208
|
continue |
|
209
|
|
|
210
|
# Match resource name against provider prefixes |
|
211
|
for prefix, provider in _PROVIDER_PREFIXES.items(): |
|
212
|
if res_name.startswith(prefix): |
|
213
|
# Find or reference the provider node |
|
214
|
provider_result = self.store.query( |
|
215
|
"MATCH (p) WHERE p.name = $provider " |
|
216
|
"AND p.semantic_type = $prov_type " |
|
217
|
"RETURN p.name", |
|
218
|
{"provider": provider, "prov_type": "terraform_provider"}, |
|
219
|
) |
|
220
|
provider_rows = provider_result.result_set or [] |
|
221
|
if provider_rows and provider_rows[0][0]: |
|
222
|
self._add_semantic_edge( |
|
223
|
res_name, |
|
224
|
"BELONGS_TO", |
|
225
|
provider, |
|
226
|
) |
|
227
|
edges_added += 1 |
|
228
|
break # Only match the first (most specific) prefix |
|
229
|
|
|
230
|
return edges_added |
|
231
|
|