template.py 41 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972
  1. from __future__ import annotations
  2. import base64
  3. import importlib
  4. import logging
  5. import os
  6. import re
  7. import secrets
  8. import string
  9. from dataclasses import dataclass, field
  10. from functools import lru_cache
  11. from pathlib import Path
  12. from typing import Any, Literal
  13. import yaml
  14. from jinja2 import Environment, FileSystemLoader, meta
  15. from jinja2.exceptions import (
  16. TemplateError as Jinja2TemplateError,
  17. )
  18. from jinja2.exceptions import (
  19. TemplateNotFound as Jinja2TemplateNotFound,
  20. )
  21. from jinja2.exceptions import (
  22. TemplateSyntaxError as Jinja2TemplateSyntaxError,
  23. )
  24. from jinja2.exceptions import (
  25. UndefinedError,
  26. )
  27. from jinja2.sandbox import SandboxedEnvironment
  28. from ..exceptions import (
  29. IncompatibleSchemaVersionError,
  30. RenderErrorContext,
  31. TemplateLoadError,
  32. TemplateRenderError,
  33. TemplateSyntaxError,
  34. TemplateValidationError,
  35. YAMLParseError,
  36. )
  37. from ..version import is_compatible
  38. from .variable_collection import VariableCollection
  39. logger = logging.getLogger(__name__)
  40. class TemplateErrorHandler:
  41. """Handles parsing and formatting of template rendering errors.
  42. This class provides utilities for:
  43. - Extracting error context from template files
  44. - Generating helpful suggestions based on Jinja2 errors
  45. - Parsing Jinja2 exceptions into structured error information
  46. """
  47. @staticmethod
  48. def extract_error_context(file_path: Path, line_number: int | None, context_size: int = 3) -> list[str]:
  49. """Extract lines of context around an error location.
  50. Args:
  51. file_path: Path to the file with the error
  52. line_number: Line number where error occurred (1-indexed)
  53. context_size: Number of lines to show before and after
  54. Returns:
  55. List of context lines with line numbers
  56. """
  57. if not line_number or not file_path.exists():
  58. return []
  59. try:
  60. with file_path.open(encoding="utf-8") as f:
  61. lines = f.readlines()
  62. start_line = max(0, line_number - context_size - 1)
  63. end_line = min(len(lines), line_number + context_size)
  64. context = []
  65. for i in range(start_line, end_line):
  66. line_num = i + 1
  67. marker = ">>>" if line_num == line_number else " "
  68. context.append(f"{marker} {line_num:4d} | {lines[i].rstrip()}")
  69. return context
  70. except OSError:
  71. return []
  72. @staticmethod
  73. def get_common_jinja_suggestions(error_msg: str, available_vars: set) -> list[str]:
  74. """Generate helpful suggestions based on common Jinja2 errors.
  75. Args:
  76. error_msg: The error message from Jinja2
  77. available_vars: Set of available variable names
  78. Returns:
  79. List of actionable suggestions
  80. """
  81. suggestions = []
  82. error_lower = error_msg.lower()
  83. # Undefined variable errors
  84. if "undefined" in error_lower or "is not defined" in error_lower:
  85. # Try to extract variable name from error message
  86. var_match = re.search(r"'([^']+)'.*is undefined", error_msg)
  87. if not var_match:
  88. var_match = re.search(r"'([^']+)'.*is not defined", error_msg)
  89. if var_match:
  90. undefined_var = var_match.group(1)
  91. suggestions.append(f"Variable '{undefined_var}' is not defined in the template spec")
  92. # Suggest similar variable names (basic fuzzy matching)
  93. similar = [
  94. v
  95. for v in available_vars
  96. if undefined_var.lower() in v.lower() or v.lower() in undefined_var.lower()
  97. ]
  98. if similar:
  99. suggestions.append(f"Did you mean one of these? {', '.join(sorted(similar)[:5])}")
  100. suggestions.append(f"Add '{undefined_var}' to your template.yaml spec with a default value")
  101. suggestions.append("Or use the Jinja2 default filter: {{ " + undefined_var + " | default('value') }}")
  102. else:
  103. suggestions.append("Check that all variables used in templates are defined in template.yaml")
  104. suggestions.append("Use the Jinja2 default filter for optional variables: {{ var | default('value') }}")
  105. # Syntax errors
  106. elif "unexpected" in error_lower or "expected" in error_lower:
  107. suggestions.append("Check for syntax errors in your Jinja2 template")
  108. suggestions.append("Common issues: missing {% endfor %}, {% endif %}, or {% endblock %}")
  109. suggestions.append("Make sure all {{ }} and {% %} tags are properly closed")
  110. # Filter errors
  111. elif "filter" in error_lower:
  112. suggestions.append("Check that the filter name is spelled correctly")
  113. suggestions.append("Verify the filter exists in Jinja2 built-in filters")
  114. suggestions.append("Make sure filter arguments are properly formatted")
  115. # Template not found
  116. elif "not found" in error_lower or "does not exist" in error_lower:
  117. suggestions.append("Check that the included/imported template file exists")
  118. suggestions.append("Verify the template path is relative to the template directory")
  119. suggestions.append("Make sure the file has the .j2 extension if it's a Jinja2 template")
  120. # Type errors
  121. elif "type" in error_lower and ("int" in error_lower or "str" in error_lower or "bool" in error_lower):
  122. suggestions.append("Check that variable values have the correct type")
  123. suggestions.append("Use Jinja2 filters to convert types: {{ var | int }}, {{ var | string }}")
  124. # Add generic helpful tip
  125. if not suggestions:
  126. suggestions.append("Check the Jinja2 template syntax and variable usage")
  127. suggestions.append("Enable --debug mode for more detailed rendering information")
  128. return suggestions
  129. @classmethod
  130. def parse_jinja_error(
  131. cls,
  132. error: Exception,
  133. template_file: TemplateFile,
  134. template_dir: Path,
  135. available_vars: set,
  136. ) -> tuple[str, int | None, int | None, list[str], list[str]]:
  137. """Parse a Jinja2 exception to extract detailed error information.
  138. Args:
  139. error: The Jinja2 exception
  140. template_file: The TemplateFile being rendered
  141. template_dir: Template directory path
  142. available_vars: Set of available variable names
  143. Returns:
  144. Tuple of (error_message, line_number, column, context_lines, suggestions)
  145. """
  146. error_msg = str(error)
  147. line_number = None
  148. column = None
  149. context_lines = []
  150. suggestions = []
  151. # Extract line number from Jinja2 errors
  152. if hasattr(error, "lineno"):
  153. line_number = error.lineno
  154. # Extract file path and get context
  155. file_path = template_dir / template_file.relative_path
  156. if line_number and file_path.exists():
  157. context_lines = cls.extract_error_context(file_path, line_number)
  158. # Generate suggestions based on error type
  159. if isinstance(error, UndefinedError):
  160. error_msg = f"Undefined variable: {error}"
  161. suggestions = cls.get_common_jinja_suggestions(str(error), available_vars)
  162. elif isinstance(error, Jinja2TemplateSyntaxError):
  163. error_msg = f"Template syntax error: {error}"
  164. suggestions = cls.get_common_jinja_suggestions(str(error), available_vars)
  165. elif isinstance(error, Jinja2TemplateNotFound):
  166. error_msg = f"Template file not found: {error}"
  167. suggestions = cls.get_common_jinja_suggestions(str(error), available_vars)
  168. else:
  169. # Generic Jinja2 error
  170. suggestions = cls.get_common_jinja_suggestions(error_msg, available_vars)
  171. return error_msg, line_number, column, context_lines, suggestions
  172. @dataclass
  173. class TemplateFile:
  174. """Represents a single file within a template directory."""
  175. relative_path: Path
  176. file_type: Literal["j2", "static"]
  177. output_path: Path # The path it will have in the output directory
  178. @dataclass
  179. class TemplateMetadata:
  180. """Represents template metadata with proper typing."""
  181. name: str
  182. description: str
  183. author: str
  184. date: str
  185. version: str
  186. module: str = ""
  187. tags: list[str] = field(default_factory=list)
  188. library: str = "unknown"
  189. library_type: str = "git" # Type of library ("git" or "static")
  190. next_steps: str = ""
  191. draft: bool = False
  192. def __init__(
  193. self,
  194. template_data: dict,
  195. library_name: str | None = None,
  196. library_type: str = "git",
  197. ) -> None:
  198. """Initialize TemplateMetadata from parsed YAML template data.
  199. Args:
  200. template_data: Parsed YAML data from template.yaml
  201. library_name: Name of the library this template belongs to
  202. """
  203. # Validate metadata format first
  204. self._validate_metadata(template_data)
  205. # Extract metadata section
  206. metadata_section = template_data.get("metadata", {})
  207. self.name = metadata_section.get("name", "")
  208. # YAML block scalar (|) preserves a trailing newline. Remove only trailing newlines
  209. # while preserving internal newlines/formatting.
  210. raw_description = metadata_section.get("description", "")
  211. # TODO: remove when all templates have been migrated to markdown
  212. description = raw_description.rstrip("\n") if isinstance(raw_description, str) else str(raw_description)
  213. self.description = description or "No description available"
  214. self.author = metadata_section.get("author", "")
  215. self.date = metadata_section.get("date", "")
  216. self.version = metadata_section.get("version", "")
  217. self.module = metadata_section.get("module", "")
  218. self.tags = metadata_section.get("tags", []) or []
  219. self.library = library_name or "unknown"
  220. self.library_type = library_type
  221. self.draft = metadata_section.get("draft", False)
  222. # Extract next_steps (optional)
  223. raw_next_steps = metadata_section.get("next_steps", "")
  224. if isinstance(raw_next_steps, str):
  225. next_steps = raw_next_steps.rstrip("\n")
  226. else:
  227. next_steps = str(raw_next_steps) if raw_next_steps else ""
  228. self.next_steps = next_steps
  229. @staticmethod
  230. def _validate_metadata(template_data: dict) -> None:
  231. """Validate that template has required 'metadata' section with all required fields.
  232. Args:
  233. template_data: Parsed YAML data from template.yaml
  234. Raises:
  235. ValueError: If metadata section is missing or incomplete
  236. """
  237. metadata_section = template_data.get("metadata")
  238. if metadata_section is None:
  239. raise ValueError("Template format error: missing 'metadata' section")
  240. # Validate that metadata section has all required fields
  241. required_fields = ["name", "author", "version", "date", "description"]
  242. missing_fields = [field for field in required_fields if not metadata_section.get(field)]
  243. if missing_fields:
  244. raise ValueError(f"Template format error: missing required metadata fields: {missing_fields}")
  245. @dataclass
  246. class Template:
  247. """Represents a template directory."""
  248. def __init__(self, template_dir: Path, library_name: str, library_type: str = "git") -> None:
  249. """Create a Template instance from a directory path.
  250. Args:
  251. template_dir: Path to the template directory
  252. library_name: Name of the library this template belongs to
  253. library_type: Type of library ("git" or "static"), defaults to "git"
  254. """
  255. logger.debug(f"Loading template from directory: {template_dir}")
  256. self.template_dir = template_dir
  257. self.id = template_dir.name
  258. self.original_id = template_dir.name # Store the original ID
  259. self.library_name = library_name
  260. self.library_type = library_type
  261. # Initialize caches for lazy loading
  262. self.__module_specs: dict | None = None
  263. self.__merged_specs: dict | None = None
  264. self.__jinja_env: Environment | None = None
  265. self.__used_variables: set[str] | None = None
  266. self.__variables: VariableCollection | None = None
  267. self.__template_files: list[TemplateFile] | None = None # New attribute
  268. self._schema_deprecation_warned: bool = False # Track if deprecation warning shown
  269. try:
  270. # Find and parse the main template file (template.yaml or template.yml)
  271. main_template_path = self._find_main_template_file()
  272. with main_template_path.open(encoding="utf-8") as f:
  273. # Load all YAML documents (handles templates with empty lines before ---)
  274. documents = list(yaml.safe_load_all(f))
  275. # Filter out None/empty documents and get the first non-empty one
  276. valid_docs = [doc for doc in documents if doc is not None]
  277. if not valid_docs:
  278. raise ValueError("Template file contains no valid YAML data")
  279. if len(valid_docs) > 1:
  280. logger.warning("Template file contains multiple YAML documents, using the first one")
  281. self._template_data = valid_docs[0]
  282. # Validate template data
  283. if not isinstance(self._template_data, dict):
  284. raise ValueError("Template file must contain a valid YAML dictionary")
  285. # Load metadata (always needed)
  286. self.metadata = TemplateMetadata(self._template_data, library_name, library_type)
  287. logger.debug(f"Loaded metadata: {self.metadata}")
  288. # Validate 'kind' field (always needed)
  289. self._validate_kind(self._template_data)
  290. # Extract schema version (default to 1.0 for backward compatibility)
  291. self.schema_version = str(self._template_data.get("schema", "1.0"))
  292. logger.debug(f"Template schema version: {self.schema_version}")
  293. # Note: Schema version validation is done by the module when loading templates
  294. # NOTE: File collection is now lazy-loaded via the template_files property
  295. # This significantly improves performance when listing many templates
  296. logger.info(f"Loaded template '{self.id}' (v{self.metadata.version})")
  297. except (ValueError, FileNotFoundError) as e:
  298. logger.error(f"Error loading template from {template_dir}: {e}")
  299. raise TemplateLoadError(f"Error loading template from {template_dir}: {e}") from e
  300. except yaml.YAMLError as e:
  301. logger.error(f"YAML parsing error in template {template_dir}: {e}")
  302. raise YAMLParseError(str(template_dir / "template.y*ml"), e) from e
  303. except OSError as e:
  304. logger.error(f"File I/O error loading template {template_dir}: {e}")
  305. raise TemplateLoadError(f"File I/O error loading template from {template_dir}: {e}") from e
  306. def set_qualified_id(self, library_name: str | None = None) -> None:
  307. """Set a qualified ID for this template (used when duplicates exist across libraries).
  308. Args:
  309. library_name: Name of the library to qualify with. If None, uses self.library_name
  310. """
  311. lib_name = library_name or self.library_name
  312. self.id = f"{self.original_id}.{lib_name}"
  313. logger.debug(f"Template ID qualified: {self.original_id} -> {self.id}")
  314. def _find_main_template_file(self) -> Path:
  315. """Find the main template file (template.yaml or template.yml)."""
  316. for filename in ["template.yaml", "template.yml"]:
  317. path = self.template_dir / filename
  318. if path.exists():
  319. return path
  320. raise FileNotFoundError(f"Main template file (template.yaml or template.yml) not found in {self.template_dir}")
  321. @staticmethod
  322. @lru_cache(maxsize=32)
  323. def _load_module_specs_for_schema(kind: str, schema_version: str) -> dict:
  324. """Load specifications from the corresponding module for a specific schema version.
  325. DEPRECATION NOTICE (v0.1.3): Module schemas are being deprecated. Templates should define
  326. all variables in their template.yaml spec section. In future versions, this method will
  327. return an empty dict and templates will need to be self-contained.
  328. Uses LRU cache to avoid re-loading the same module spec multiple times.
  329. This significantly improves performance when listing many templates of the same kind.
  330. Args:
  331. kind: The module kind (e.g., 'compose', 'terraform')
  332. schema_version: The schema version to load (e.g., '1.0', '1.1')
  333. Returns:
  334. Dictionary containing the module's spec for the requested schema version,
  335. or empty dict if kind is empty
  336. Raises:
  337. ValueError: If module cannot be loaded or spec is invalid
  338. """
  339. if not kind:
  340. return {}
  341. # Log cache statistics for performance monitoring
  342. cache_info = Template._load_module_specs_for_schema.cache_info()
  343. logger.debug(
  344. f"Loading module spec: kind='{kind}', schema={schema_version} "
  345. f"(cache: hits={cache_info.hits}, misses={cache_info.misses}, size={cache_info.currsize})"
  346. )
  347. try:
  348. module = importlib.import_module(f"cli.modules.{kind}")
  349. # Check if module has schema-specific specs (multi-schema support)
  350. # Try SCHEMAS constant first (uppercase), then schemas attribute
  351. schemas = getattr(module, "SCHEMAS", None) or getattr(module, "schemas", None)
  352. if schemas and schema_version in schemas:
  353. spec = schemas[schema_version]
  354. logger.debug(f"Loaded and cached module spec for kind '{kind}' schema {schema_version}")
  355. else:
  356. # Fallback to default spec if schema mapping not available
  357. spec = getattr(module, "spec", {})
  358. logger.debug(f"Loaded and cached module spec for kind '{kind}' (default/no schema mapping)")
  359. return spec
  360. except Exception as e:
  361. raise ValueError(f"Error loading module specifications for kind '{kind}': {e}") from e
  362. def _merge_specs(self, module_specs: dict, template_specs: dict) -> dict:
  363. """Deep merge template specs with module specs using VariableCollection.
  364. Uses VariableCollection's native merge() method for consistent merging logic.
  365. Module specs are base, template specs override with origin tracking.
  366. """
  367. # Create VariableCollection from module specs (base)
  368. module_collection = VariableCollection(module_specs) if module_specs else VariableCollection({})
  369. # Set origin for module variables
  370. for section in module_collection.get_sections().values():
  371. for variable in section.variables.values():
  372. if not variable.origin:
  373. variable.origin = "module"
  374. # Merge template specs into module specs (template overrides)
  375. if template_specs:
  376. merged_collection = module_collection.merge(template_specs, origin="template")
  377. else:
  378. merged_collection = module_collection
  379. # Convert back to dict format
  380. merged_spec = {}
  381. for section_key, section in merged_collection.get_sections().items():
  382. merged_spec[section_key] = section.to_dict()
  383. return merged_spec
  384. def _warn_about_inherited_variables(self, module_specs: dict, template_specs: dict) -> None:
  385. """Warn about variables inherited from module schemas (deprecation warning).
  386. DEPRECATION (v0.1.3): Templates should define all variables explicitly in template.yaml.
  387. This method detects variables that are used in template files but come from module schemas,
  388. and warns users to add them to the template spec.
  389. Args:
  390. module_specs: Variables defined in module schema
  391. template_specs: Variables defined in template.yaml
  392. """
  393. # Skip if no module specs (already self-contained)
  394. if not module_specs:
  395. return
  396. # Collect variables from module specs
  397. module_vars = set()
  398. for section_data in module_specs.values():
  399. if isinstance(section_data, dict) and "vars" in section_data:
  400. module_vars.update(section_data["vars"].keys())
  401. # Collect variables explicitly defined in template
  402. template_vars = set()
  403. for section_data in (template_specs or {}).values():
  404. if isinstance(section_data, dict) and "vars" in section_data:
  405. template_vars.update(section_data["vars"].keys())
  406. # Get variables actually used in template files
  407. used_vars = self.used_variables
  408. # Find variables that are:
  409. # 1. Used in template files AND defined in module schema (inherited variables)
  410. # 2. NOT explicitly defined in template.yaml (missing definitions)
  411. # These are the problematic ones - the template relies on schema
  412. inherited_vars = used_vars & module_vars
  413. missing_definitions = inherited_vars - template_vars
  414. # Only warn once per template (use a flag to avoid repeated warnings)
  415. if missing_definitions and not self._schema_deprecation_warned:
  416. self._schema_deprecation_warned = True
  417. # Show first N variables in warning, full list in debug
  418. max_shown_vars = 10
  419. shown_vars = sorted(list(missing_definitions)[:max_shown_vars])
  420. ellipsis = "..." if len(missing_definitions) > max_shown_vars else ""
  421. logger.warning(
  422. f"DEPRECATION WARNING: Template '{self.id}' uses {len(missing_definitions)} "
  423. f"variable(s) from module schema without defining them in template.yaml. "
  424. f"In future versions, all used variables must be defined in template.yaml. "
  425. f"Missing definitions: {', '.join(shown_vars)}{ellipsis}"
  426. )
  427. logger.debug(
  428. f"Template '{self.id}' should add these variable definitions to template.yaml spec: "
  429. f"{sorted(missing_definitions)}"
  430. )
  431. def _collect_template_files(self) -> None:
  432. """Collects all TemplateFile objects in the template directory."""
  433. template_files: list[TemplateFile] = []
  434. for root, _, files in os.walk(self.template_dir):
  435. for filename in files:
  436. file_path = Path(root) / filename
  437. relative_path = file_path.relative_to(self.template_dir)
  438. # Skip the main template file
  439. if filename in ["template.yaml", "template.yml"]:
  440. continue
  441. if filename.endswith(".j2"):
  442. file_type: Literal["j2", "static"] = "j2"
  443. output_path = relative_path.with_suffix("") # Remove .j2 suffix
  444. else:
  445. file_type = "static"
  446. output_path = relative_path # Static files keep their name
  447. template_files.append(
  448. TemplateFile(
  449. relative_path=relative_path,
  450. file_type=file_type,
  451. output_path=output_path,
  452. )
  453. )
  454. self.__template_files = template_files
  455. def _extract_all_used_variables(self) -> set[str]:
  456. """Extract all undeclared variables from all .j2 files in the template directory.
  457. Raises:
  458. ValueError: If any Jinja2 template has syntax errors
  459. """
  460. used_variables: set[str] = set()
  461. syntax_errors = []
  462. # Track which file uses which variable (for debugging)
  463. self._variable_usage_map: dict[str, list[str]] = {}
  464. for template_file in self.template_files: # Iterate over TemplateFile objects
  465. if template_file.file_type == "j2":
  466. file_path = self.template_dir / template_file.relative_path
  467. try:
  468. with file_path.open(encoding="utf-8") as f:
  469. content = f.read()
  470. ast = self.jinja_env.parse(content) # Use lazy-loaded jinja_env
  471. file_vars = meta.find_undeclared_variables(ast)
  472. used_variables.update(file_vars)
  473. # Track which file uses each variable
  474. for var in file_vars:
  475. if var not in self._variable_usage_map:
  476. self._variable_usage_map[var] = []
  477. self._variable_usage_map[var].append(str(template_file.relative_path))
  478. except OSError as e:
  479. relative_path = file_path.relative_to(self.template_dir)
  480. syntax_errors.append(f" - {relative_path}: File I/O error: {e}")
  481. except Exception as e:
  482. # Collect syntax errors for Jinja2 issues
  483. relative_path = file_path.relative_to(self.template_dir)
  484. syntax_errors.append(f" - {relative_path}: {e}")
  485. # Raise error if any syntax errors were found
  486. if syntax_errors:
  487. logger.error(f"Jinja2 syntax errors found in template '{self.id}'")
  488. raise TemplateSyntaxError(self.id, syntax_errors)
  489. return used_variables
  490. def _filter_specs_to_used(
  491. self,
  492. used_variables: set,
  493. merged_specs: dict,
  494. _module_specs: dict,
  495. template_specs: dict,
  496. ) -> dict:
  497. """Filter specs to only include variables used in templates using VariableCollection.
  498. Uses VariableCollection's native filter_to_used() method.
  499. Keeps sensitive variables only if they're defined in the template spec or actually used.
  500. """
  501. # Build set of variables explicitly defined in template spec
  502. template_defined_vars = set()
  503. for section_data in (template_specs or {}).values():
  504. if isinstance(section_data, dict) and "vars" in section_data:
  505. template_defined_vars.update(section_data["vars"].keys())
  506. # Create VariableCollection from merged specs
  507. merged_collection = VariableCollection(merged_specs)
  508. # Filter to only used variables (and sensitive ones that are template-defined)
  509. # We keep sensitive variables that are either:
  510. # 1. Actually used in template files, OR
  511. # 2. Explicitly defined in the template spec (even if not yet used)
  512. variables_to_keep = used_variables | template_defined_vars
  513. filtered_collection = merged_collection.filter_to_used(variables_to_keep, keep_sensitive=False)
  514. # Convert back to dict format
  515. filtered_specs = {}
  516. for section_key, section in filtered_collection.get_sections().items():
  517. filtered_specs[section_key] = section.to_dict()
  518. return filtered_specs
  519. def _validate_schema_version(self, module_schema: str, module_name: str) -> None:
  520. """Validate that template schema version is supported by the module.
  521. Args:
  522. module_schema: Schema version supported by the module
  523. module_name: Name of the module (for error messages)
  524. Raises:
  525. IncompatibleSchemaVersionError: If template schema > module schema
  526. """
  527. template_schema = self.schema_version
  528. # Compare schema versions
  529. if not is_compatible(module_schema, template_schema):
  530. logger.error(
  531. f"Template '{self.id}' uses schema version {template_schema}, "
  532. f"but module '{module_name}' only supports up to {module_schema}"
  533. )
  534. raise IncompatibleSchemaVersionError(
  535. template_id=self.id,
  536. template_schema=template_schema,
  537. module_schema=module_schema,
  538. module_name=module_name,
  539. )
  540. logger.debug(
  541. f"Template '{self.id}' schema version compatible: "
  542. f"template uses {template_schema}, module supports {module_schema}"
  543. )
  544. @staticmethod
  545. def _validate_kind(template_data: dict) -> None:
  546. """Validate that template has required 'kind' field.
  547. Args:
  548. template_data: Parsed YAML data from template.yaml
  549. Raises:
  550. ValueError: If 'kind' field is missing
  551. """
  552. if not template_data.get("kind"):
  553. raise TemplateValidationError("Template format error: missing 'kind' field")
  554. def _validate_variable_definitions(self, used_variables: set[str], merged_specs: dict[str, Any]) -> None:
  555. """Validate that all variables used in Jinja2 content are defined in the spec."""
  556. defined_variables = set()
  557. for section_data in merged_specs.values():
  558. if "vars" in section_data and isinstance(section_data["vars"], dict):
  559. defined_variables.update(section_data["vars"].keys())
  560. undefined_variables = used_variables - defined_variables
  561. if undefined_variables:
  562. undefined_list = sorted(undefined_variables)
  563. # Build file location info for each undefined variable
  564. file_locations = []
  565. for var_name in undefined_list:
  566. if hasattr(self, "_variable_usage_map") and var_name in self._variable_usage_map:
  567. files = self._variable_usage_map[var_name]
  568. file_locations.append(f" • {var_name}: {', '.join(files)}")
  569. error_msg = (
  570. f"Template validation error in '{self.id}': "
  571. f"Variables used in template content but not defined in spec:\n"
  572. )
  573. if file_locations:
  574. error_msg += "\n".join(file_locations) + "\n"
  575. else:
  576. error_msg += f"{undefined_list}\n"
  577. error_msg += (
  578. "\nPlease add these variables to your template's template.yaml spec. "
  579. "Each variable must have a default value.\n\n"
  580. "Example:\n"
  581. "spec:\n"
  582. " general:\n"
  583. " vars:\n"
  584. )
  585. for var_name in undefined_list:
  586. error_msg += (
  587. f" {var_name}:\n"
  588. f" type: str\n"
  589. f" description: Description for {var_name}\n"
  590. f" default: <your_default_value_here>\n"
  591. )
  592. # Only log at DEBUG level - the exception will be displayed to user
  593. logger.debug(error_msg)
  594. raise TemplateValidationError(error_msg)
  595. @staticmethod
  596. def _create_jinja_env(searchpath: Path) -> Environment:
  597. """Create sandboxed Jinja2 environment for secure template processing.
  598. Uses SandboxedEnvironment to prevent code injection vulnerabilities
  599. when processing untrusted templates. This restricts access to dangerous
  600. operations while still allowing safe template rendering.
  601. Returns:
  602. SandboxedEnvironment configured for template processing.
  603. """
  604. # NOTE Use SandboxedEnvironment for security - prevents arbitrary code execution
  605. return SandboxedEnvironment(
  606. loader=FileSystemLoader(searchpath),
  607. trim_blocks=True,
  608. lstrip_blocks=True,
  609. keep_trailing_newline=False,
  610. )
  611. def _generate_autogenerated_values(self, variables: VariableCollection, variable_values: dict) -> None:
  612. """Generate values for autogenerated variables that are empty.
  613. Supports both plain and base64-encoded autogenerated values based on
  614. the autogenerated_base64 flag. Base64 encoding generates random bytes
  615. and encodes them, which is more suitable for cryptographic keys.
  616. """
  617. for section in variables.get_sections().values():
  618. for var_name, variable in section.variables.items():
  619. if variable.autogenerated and (variable.value is None or variable.value == ""):
  620. length = getattr(variable, "autogenerated_length", 32)
  621. use_base64 = getattr(variable, "autogenerated_base64", False)
  622. if use_base64:
  623. # Generate random bytes and base64 encode
  624. # Note: length refers to number of random bytes, not base64 string length
  625. random_bytes = secrets.token_bytes(length)
  626. generated_value = base64.b64encode(random_bytes).decode("utf-8")
  627. logger.debug(
  628. f"Auto-generated base64 value for variable '{var_name}' "
  629. f"(bytes: {length}, encoded length: {len(generated_value)})"
  630. )
  631. else:
  632. # Generate alphanumeric string
  633. alphabet = string.ascii_letters + string.digits
  634. generated_value = "".join(secrets.choice(alphabet) for _ in range(length))
  635. logger.debug(f"Auto-generated value for variable '{var_name}' (length: {length})")
  636. variable_values[var_name] = generated_value
  637. def _log_render_start(self, debug: bool, variable_values: dict) -> None:
  638. """Log rendering start information."""
  639. if debug:
  640. logger.info(f"Rendering template '{self.id}' in debug mode")
  641. logger.info(f"Available variables: {sorted(variable_values.keys())}")
  642. logger.info(f"Variable values: {variable_values}")
  643. else:
  644. logger.debug(f"Rendering template '{self.id}' with variables: {variable_values}")
  645. def _render_jinja2_file(self, template_file, variable_values: dict, _available_vars: set, debug: bool) -> str:
  646. """Render a single Jinja2 template file."""
  647. if debug:
  648. logger.info(f"Rendering Jinja2 template: {template_file.relative_path}")
  649. template = self.jinja_env.get_template(str(template_file.relative_path))
  650. rendered_content = template.render(**variable_values)
  651. rendered_content = self._sanitize_content(rendered_content, template_file.output_path)
  652. if debug:
  653. logger.info(f"Successfully rendered: {template_file.relative_path} -> {template_file.output_path}")
  654. return rendered_content
  655. def _handle_jinja2_error(
  656. self,
  657. e: Exception,
  658. template_file,
  659. available_vars: set,
  660. variable_values: dict,
  661. debug: bool,
  662. ) -> None:
  663. """Handle Jinja2 rendering errors."""
  664. error_msg, line_num, col, context_lines, suggestions = TemplateErrorHandler.parse_jinja_error(
  665. e, template_file, self.template_dir, available_vars
  666. )
  667. logger.error(f"Error rendering template file {template_file.relative_path}: {error_msg}")
  668. context = RenderErrorContext(
  669. file_path=str(template_file.relative_path),
  670. line_number=line_num,
  671. column=col,
  672. context_lines=context_lines,
  673. variable_context={k: str(v) for k, v in variable_values.items()} if debug else {},
  674. suggestions=suggestions,
  675. original_error=e,
  676. )
  677. raise TemplateRenderError(message=error_msg, context=context) from e
  678. def _render_static_file(self, template_file, debug: bool) -> str:
  679. """Read and return content of a static file."""
  680. file_path = self.template_dir / template_file.relative_path
  681. if debug:
  682. logger.info(f"Copying static file: {template_file.relative_path}")
  683. try:
  684. with file_path.open(encoding="utf-8") as f:
  685. return f.read()
  686. except OSError as e:
  687. logger.error(f"Error reading static file {file_path}: {e}")
  688. context = RenderErrorContext(
  689. file_path=str(template_file.relative_path),
  690. suggestions=["Check that the file exists and has read permissions"],
  691. original_error=e,
  692. )
  693. raise TemplateRenderError(
  694. message=f"Error reading static file: {e}",
  695. context=context,
  696. ) from e
  697. def render(self, variables: VariableCollection, debug: bool = False) -> tuple[dict[str, str], dict[str, Any]]:
  698. """Render all .j2 files in the template directory.
  699. Args:
  700. variables: VariableCollection with values to use for rendering
  701. debug: Enable debug mode with verbose output
  702. Returns:
  703. Tuple of (rendered_files, variable_values) where variable_values includes autogenerated values.
  704. Empty files (files with only whitespace) are excluded from the returned dict.
  705. """
  706. variable_values = variables.get_satisfied_values()
  707. self._generate_autogenerated_values(variables, variable_values)
  708. self._log_render_start(debug, variable_values)
  709. rendered_files = {}
  710. skipped_files = []
  711. available_vars = set(variable_values.keys())
  712. for template_file in self.template_files:
  713. if template_file.file_type == "j2":
  714. try:
  715. content = self._render_jinja2_file(template_file, variable_values, available_vars, debug)
  716. # Skip empty files (only whitespace, empty string, or just YAML document separator)
  717. stripped = content.strip()
  718. if stripped and stripped != "---":
  719. rendered_files[str(template_file.output_path)] = content
  720. else:
  721. skipped_files.append(str(template_file.output_path))
  722. if debug:
  723. logger.info(f"Skipping empty file: {template_file.output_path}")
  724. except (
  725. UndefinedError,
  726. Jinja2TemplateSyntaxError,
  727. Jinja2TemplateNotFound,
  728. Jinja2TemplateError,
  729. ) as e:
  730. self._handle_jinja2_error(e, template_file, available_vars, variable_values, debug)
  731. except Exception as e:
  732. logger.error(f"Unexpected error rendering template file {template_file.relative_path}: {e}")
  733. context = RenderErrorContext(
  734. file_path=str(template_file.relative_path),
  735. suggestions=["This is an unexpected error. Please check the template for issues."],
  736. original_error=e,
  737. )
  738. raise TemplateRenderError(
  739. message=f"Unexpected rendering error: {e}",
  740. context=context,
  741. ) from e
  742. elif template_file.file_type == "static":
  743. content = self._render_static_file(template_file, debug)
  744. # Static files are always included, even if empty
  745. rendered_files[str(template_file.output_path)] = content
  746. if skipped_files:
  747. logger.debug(f"Skipped {len(skipped_files)} empty file(s): {', '.join(skipped_files)}")
  748. return rendered_files, variable_values
  749. def _sanitize_content(self, content: str, _file_path: Path) -> str:
  750. """Sanitize rendered content by removing excessive blank lines and trailing whitespace."""
  751. if not content:
  752. return content
  753. lines = [line.rstrip() for line in content.split("\n")]
  754. sanitized = []
  755. prev_blank = False
  756. for line in lines:
  757. is_blank = not line
  758. if is_blank and prev_blank:
  759. continue # Skip consecutive blank lines
  760. sanitized.append(line)
  761. prev_blank = is_blank
  762. # Remove leading blanks and ensure single trailing newline
  763. return "\n".join(sanitized).lstrip("\n").rstrip("\n") + "\n"
  764. @property
  765. def template_files(self) -> list[TemplateFile]:
  766. if self.__template_files is None:
  767. self._collect_template_files() # Populate self.__template_files
  768. return self.__template_files
  769. @property
  770. def template_specs(self) -> dict:
  771. """Get the spec section from template YAML data."""
  772. return self._template_data.get("spec", {})
  773. @property
  774. def module_specs(self) -> dict:
  775. """Get the spec from the module definition for this template's schema version."""
  776. if self.__module_specs is None:
  777. kind = self._template_data.get("kind")
  778. self.__module_specs = self._load_module_specs_for_schema(kind, self.schema_version)
  779. return self.__module_specs
  780. @property
  781. def merged_specs(self) -> dict:
  782. if self.__merged_specs is None:
  783. # Warn about inherited variables (deprecation)
  784. self._warn_about_inherited_variables(self.module_specs, self.template_specs)
  785. self.__merged_specs = self._merge_specs(self.module_specs, self.template_specs)
  786. return self.__merged_specs
  787. @property
  788. def jinja_env(self) -> Environment:
  789. if self.__jinja_env is None:
  790. self.__jinja_env = self._create_jinja_env(self.template_dir)
  791. return self.__jinja_env
  792. @property
  793. def used_variables(self) -> set[str]:
  794. if self.__used_variables is None:
  795. self.__used_variables = self._extract_all_used_variables()
  796. return self.__used_variables
  797. @property
  798. def variables(self) -> VariableCollection:
  799. if self.__variables is None:
  800. # Validate that all used variables are defined
  801. self._validate_variable_definitions(self.used_variables, self.merged_specs)
  802. # Filter specs to only used variables
  803. filtered_specs = self._filter_specs_to_used(
  804. self.used_variables,
  805. self.merged_specs,
  806. self.module_specs,
  807. self.template_specs,
  808. )
  809. self.__variables = VariableCollection(filtered_specs)
  810. # Sort sections: required first, then enabled, then disabled
  811. self.__variables.sort_sections()
  812. return self.__variables