scripts/ci/prek/check_security_doc_constants.py - airflow - Git at Google

 #!/usr/bin/env python
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information
 # regarding copyright ownership.  The ASF licenses this file
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
 #
 #   http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
 # /// script
 # requires-python = ">=3.10,<3.11"
 # dependencies = [
 #   "pyyaml>=6.0.3",
 #   "rich>=13.6.0",
 # ]
 # ///
 """
 Validate and auto-update security documentation against config.yml.

 Checks performed:
   1. Every ``[section] option`` reference in the security RST files corresponds to an
      actual option in config.yml or provider.yaml.
   2. Default values quoted in the docs match the defaults in config.yml.
   3. Auto-updates the sensitive-variable tables in security_model.rst between
      AUTOGENERATED markers to stay in sync with config.yml and provider.yaml.
 """

 from __future__ import annotations

 import re
 import sys
 from pathlib import Path

 import yaml
 from rich.console import Console

 sys.path.insert(0, str(Path(__file__).parent.resolve()))

 from common_prek_utils import AIRFLOW_ROOT_PATH

 console = Console(color_system="standard", width=200)

 CONFIG_YML = AIRFLOW_ROOT_PATH / "airflow-core" / "src" / "airflow" / "config_templates" / "config.yml"
 PROVIDERS_ROOT = AIRFLOW_ROOT_PATH / "providers"
 SECURITY_MODEL_RST = AIRFLOW_ROOT_PATH / "airflow-core" / "docs" / "security" / "security_model.rst"

 SECURITY_DOCS = [
     AIRFLOW_ROOT_PATH / "airflow-core" / "docs" / "security" / "jwt_token_authentication.rst",
     SECURITY_MODEL_RST,
 ]

 # Pattern to match ``[section] option_name`` references in RST
 SECTION_OPTION_RE = re.compile(r"``\[(\w+)\]\s+(\w+)``")

 # Pattern to match AIRFLOW__SECTION__OPTION env var references
 ENV_VAR_RE = re.compile(r"``(AIRFLOW__\w+)``")

 # Map section+option to the AIRFLOW__ env var form
 SECTION_OPT_TO_ENV = re.compile(r"AIRFLOW__([A-Z_]+)__([A-Z_]+)")

 # Markers for autogenerated sections
 CORE_START = "   .. START AUTOGENERATED CORE SENSITIVE VARS"
 CORE_END = "   .. END AUTOGENERATED CORE SENSITIVE VARS"
 PROVIDER_START = "   .. START AUTOGENERATED PROVIDER SENSITIVE VARS"
 PROVIDER_END = "   .. END AUTOGENERATED PROVIDER SENSITIVE VARS"

 # Which components need which core config sections/options.
 # Maps (section, option) -> list of component names.
 # This is the source of truth for the "Needed by" column.
 CORE_COMPONENT_MAP: dict[tuple[str, str], str] = {
     ("api", "secret_key"): "API Server, Scheduler, Workers, Triggerer",
     ("api_auth", "jwt_secret"): "API Server, Scheduler",
     ("core", "asset_manager_kwargs"): "Dag File Processor",
     ("core", "fernet_key"): "API Server, Scheduler, Workers, Dag File Processor, Triggerer",
     ("database", "sql_alchemy_conn"): "API Server, Scheduler, Dag File Processor, Triggerer",
     ("database", "sql_alchemy_conn_async"): "API Server, Scheduler, Dag File Processor, Triggerer",
     ("database", "sql_alchemy_engine_args"): "API Server, Scheduler, Dag File Processor, Triggerer",
     ("logging", "remote_task_handler_kwargs"): "Scheduler, Workers, Triggerer",
     ("secrets", "backend_kwargs"): "Scheduler, Dag File Processor, Triggerer",
     ("sentry", "sentry_dsn"): "Scheduler, Triggerer",
     ("workers", "secrets_backend_kwargs"): "Workers",
 }

 # Human-readable descriptions for core sensitive vars
 CORE_DESCRIPTIONS: dict[tuple[str, str], str] = {
     ("api", "secret_key"): "API secret key for log token signing",
     ("api_auth", "jwt_secret"): "JWT signing key (symmetric mode)",
     ("core", "asset_manager_kwargs"): "Asset manager credentials",
     ("core", "fernet_key"): "Fernet encryption key for connections/variables at rest",
     ("database", "sql_alchemy_conn"): "Metadata database connection string",
     ("database", "sql_alchemy_conn_async"): "Async metadata database connection string",
     ("database", "sql_alchemy_engine_args"): "SQLAlchemy engine parameters (may contain credentials)",
     ("logging", "remote_task_handler_kwargs"): "Remote logging handler credentials",
     ("secrets", "backend_kwargs"): "Secrets backend credentials (non-worker mode)",
     ("sentry", "sentry_dsn"): "Sentry error reporting endpoint",
     ("workers", "secrets_backend_kwargs"): "Worker-specific secrets backend credentials",
 }


 def option_to_env_var(section: str, option: str) -> str:
     """Convert a config section+option to its AIRFLOW__ env var form."""
     return f"AIRFLOW__{section.upper()}__{option.upper()}"


 def load_core_config() -> dict:
     """Load the core config.yml."""
     with open(CONFIG_YML) as f:
         return yaml.safe_load(f)


 def load_provider_configs() -> dict[str, dict]:
     """Load provider.yaml files. Returns {provider_name: config_sections}."""
     result = {}
     for provider_yaml in sorted(PROVIDERS_ROOT.glob("*/provider.yaml")):
         with open(provider_yaml) as f:
             data = yaml.safe_load(f)
         if data and "config" in data:
             provider_name = provider_yaml.parent.name
             result[provider_name] = data["config"]
     return result


 def get_all_options(core_config: dict, provider_configs: dict[str, dict]) -> dict[tuple[str, str], dict]:
     """Return a dict of (section, option) -> option_config for all config options."""
     result = {}
     for section_name, section_data in core_config.items():
         if not isinstance(section_data, dict) or "options" not in section_data:
             continue
         for option_name, option_config in section_data["options"].items():
             if isinstance(option_config, dict):
                 result[(section_name, option_name)] = option_config

     for _provider_name, sections in provider_configs.items():
         for section_name, section_data in sections.items():
             if not isinstance(section_data, dict) or "options" not in section_data:
                 continue
             for option_name, option_config in section_data["options"].items():
                 if isinstance(option_config, dict):
                     result[(section_name, option_name)] = option_config

     return result


 def get_core_sensitive_vars(core_config: dict) -> list[tuple[str, str]]:
     """Return sorted list of (section, option) for core sensitive config options."""
     result = []
     for section_name, section_data in core_config.items():
         if not isinstance(section_data, dict) or "options" not in section_data:
             continue
         for option_name, option_config in section_data["options"].items():
             if isinstance(option_config, dict) and option_config.get("sensitive"):
                 result.append((section_name, option_name))
     return sorted(result, key=lambda x: option_to_env_var(*x))


 def get_provider_sensitive_vars(
     provider_configs: dict[str, dict],
 ) -> list[tuple[str, str, str]]:
     """Return sorted list of (provider, section, option) for provider sensitive config options."""
     result = []
     for provider_name, sections in provider_configs.items():
         for section_name, section_data in sections.items():
             if not isinstance(section_data, dict) or "options" not in section_data:
                 continue
             for option_name, option_config in section_data["options"].items():
                 if isinstance(option_config, dict) and option_config.get("sensitive"):
                     result.append((provider_name, section_name, option_name))
     return sorted(result, key=lambda x: option_to_env_var(x[1], x[2]))


 def generate_core_table(core_sensitive: list[tuple[str, str]]) -> list[str]:
     """Generate RST list-table lines for core sensitive vars."""
     lines = [
         "",
         "   **Core Airflow sensitive configuration variables:**",
         "",
         "   .. list-table::",
         "      :header-rows: 1",
         "      :widths: 40 30 30",
         "",
         "      * - Environment variable",
         "        - Description",
         "        - Needed by",
     ]
     for section, option in core_sensitive:
         env_var = option_to_env_var(section, option)
         desc = CORE_DESCRIPTIONS.get((section, option), f"[{section}] {option}")
         needed_by = CORE_COMPONENT_MAP.get((section, option), "Review per deployment")
         lines.append(f"      * - ``{env_var}``")
         lines.append(f"        - {desc}")
         lines.append(f"        - {needed_by}")

     # Check for unmapped vars and warn
     for section, option in core_sensitive:
         if (section, option) not in CORE_COMPONENT_MAP:
             console.print(
                 f"  [yellow]⚠[/] New core sensitive var [{section}] {option} — "
                 f"add it to CORE_COMPONENT_MAP in check_security_doc_constants.py"
             )
         if (section, option) not in CORE_DESCRIPTIONS:
             console.print(
                 f"  [yellow]⚠[/] New core sensitive var [{section}] {option} — "
                 f"add a description to CORE_DESCRIPTIONS in check_security_doc_constants.py"
             )

     return lines


 def generate_provider_table(provider_sensitive: list[tuple[str, str, str]]) -> list[str]:
     """Generate RST list-table lines for provider sensitive vars."""
     lines = [
         "",
         "   **Provider-specific sensitive configuration variables:**",
         "",
         "   The following variables are defined by Airflow providers and should only be set on components where",
         "   the corresponding provider functionality is needed. The decision of which components require these",
         "   variables depends on the Deployment Manager's choices about which providers and features are",
         "   enabled in each component.",
         "",
         "   .. list-table::",
         "      :header-rows: 1",
         "      :widths: 40 30 30",
         "",
         "      * - Environment variable",
         "        - Provider",
         "        - Description",
     ]
     for provider, section, option in provider_sensitive:
         env_var = option_to_env_var(section, option)
         # Generate a reasonable description from the option name
         desc = option.replace("_", " ").capitalize()
         lines.append(f"      * - ``{env_var}``")
         lines.append(f"        - {provider}")
         lines.append(f"        - {desc}")

     return lines


 def update_autogenerated_section(
     content: str, start_marker: str, end_marker: str, new_lines: list[str]
 ) -> str:
     """Replace content between markers with new content."""
     lines = content.splitlines()
     start_idx = None
     end_idx = None

     for i, line in enumerate(lines):
         if start_marker in line:
             start_idx = i
         elif end_marker in line:
             end_idx = i
             break

     if start_idx is None or end_idx is None:
         console.print(f"  [red]✗[/] Could not find markers {start_marker!r} / {end_marker!r}")
         return content

     result = lines[: start_idx + 1] + new_lines + [""] + lines[end_idx:]
     return "\n".join(result) + "\n"


 def update_sensitive_var_tables(
     core_sensitive: list[tuple[str, str]],
     provider_sensitive: list[tuple[str, str, str]],
 ) -> bool:
     """Update the autogenerated tables in security_model.rst. Returns True if changed."""
     content = SECURITY_MODEL_RST.read_text()
     original = content

     core_lines = generate_core_table(core_sensitive)
     content = update_autogenerated_section(content, CORE_START, CORE_END, core_lines)

     provider_lines = generate_provider_table(provider_sensitive)
     content = update_autogenerated_section(content, PROVIDER_START, PROVIDER_END, provider_lines)

     if content != original:
         SECURITY_MODEL_RST.write_text(content)
         return True
     return False


 def check_option_references(doc_path: Path, all_options: dict[tuple[str, str], dict]) -> list[str]:
     """Check that all [section] option references in the doc exist in config.yml."""
     errors = []
     content = doc_path.read_text()

     for line_num, line in enumerate(content.splitlines(), 1):
         for match in SECTION_OPTION_RE.finditer(line):
             section = match.group(1)
             option = match.group(2)
             if (section, option) not in all_options:
                 section_exists = any(s == section for s, _ in all_options)
                 if section_exists:
                     errors.append(
                         f"{doc_path.name}:{line_num}: Option ``[{section}] {option}`` not found in config.yml"
                     )
                 else:
                     errors.append(
                         f"{doc_path.name}:{line_num}: Section ``[{section}]`` not found in config.yml"
                     )
     return errors


 def check_env_var_references(doc_path: Path, all_options: dict[tuple[str, str], dict]) -> list[str]:
     """Check that AIRFLOW__X__Y env var references correspond to real config options."""
     errors = []
     content = doc_path.read_text()

     for line_num, line in enumerate(content.splitlines(), 1):
         # Skip lines inside autogenerated sections — those are managed by the update logic
         if "AUTOGENERATED" in line:
             continue
         for match in ENV_VAR_RE.finditer(line):
             env_var = match.group(1)
             m = SECTION_OPT_TO_ENV.match(env_var)
             if not m:
                 continue
             section = m.group(1).lower()
             option = m.group(2).lower()
             if (section, option) not in all_options:
                 section_exists = any(s == section for s, _ in all_options)
                 if section_exists:
                     errors.append(
                         f"{doc_path.name}:{line_num}: Env var ``{env_var}`` references "
                         f"option [{section}] {option} which is not in config.yml"
                     )
                 else:
                     errors.append(
                         f"{doc_path.name}:{line_num}: Env var ``{env_var}`` references "
                         f"section [{section}] which is not in config.yml"
                     )
     return errors


 def check_defaults_in_tables(doc_path: Path, all_options: dict[tuple[str, str], dict]) -> list[str]:
     """Check default values in RST table rows match config.yml."""
     errors = []
     content = doc_path.read_text()
     lines = content.splitlines()

     i = 0
     while i < len(lines):
         line = lines[i]
         match = SECTION_OPTION_RE.search(line)
         if match and "* -" in line:
             section = match.group(1)
             option = match.group(2)
             j = i + 1
             while j < len(lines) and not lines[j].strip():
                 j += 1
             if j < len(lines) and lines[j].strip().startswith("-"):
                 value_line = lines[j].strip().lstrip("- ").strip()
                 key = (section, option)
                 if key in all_options:
                     config_default = str(all_options[key].get("default", "~"))
                     doc_value = value_line.split()[0] if value_line else ""
                     doc_value = doc_value.strip("`")
                     config_default_clean = config_default.strip('"').strip("'")
                     if (
                         doc_value
                         and config_default_clean
                         and config_default_clean not in ("~", "None", "none", "")
                         and doc_value != config_default_clean
                         and not doc_value.startswith("Auto")
                         and not doc_value.startswith("None")
                         and doc_value != "``GUESS``"
                     ):
                         errors.append(
                             f"{doc_path.name}:{j + 1}: Default for [{section}] {option} is "
                             f"'{doc_value}' in docs but '{config_default_clean}' in config.yml"
                         )
         i += 1

     return errors


 def main() -> int:
     core_config = load_core_config()
     provider_configs = load_provider_configs()
     all_options = get_all_options(core_config, provider_configs)

     # Step 1: Auto-update the sensitive var tables
     core_sensitive = get_core_sensitive_vars(core_config)
     provider_sensitive = get_provider_sensitive_vars(provider_configs)

     if update_sensitive_var_tables(core_sensitive, provider_sensitive):
         console.print(
             "  [yellow]⚠[/] security_model.rst sensitive variable tables were out of date and have been updated."
         )
         console.print("  [yellow]  Please review and commit the changes.[/]")

     # Step 2: Validate references (re-read after potential update)
     all_errors: list[str] = []

     for doc_path in SECURITY_DOCS:
         if not doc_path.exists():
             console.print(f"  [yellow]⚠[/] {doc_path.name} not found, skipping")
             continue
         all_errors.extend(check_option_references(doc_path, all_options))
         all_errors.extend(check_env_var_references(doc_path, all_options))
         all_errors.extend(check_defaults_in_tables(doc_path, all_options))

     if all_errors:
         console.print()
         for error in all_errors:
             console.print(f"  [red]✗[/] {error}")
         console.print()
         console.print(f"[red]Security doc constants check failed with {len(all_errors)} error(s).[/]")
         console.print(
             "[yellow]Fix the documentation to match config.yml, or update config.yml if the docs are correct.[/]"
         )
         return 1

     console.print("[green]Security doc constants check passed.[/]")
     return 0


 if __name__ == "__main__":
     sys.exit(main())
	#!/usr/bin/env python
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.
	# /// script
	# requires-python = ">=3.10,<3.11"
	# dependencies = [
	# "pyyaml>=6.0.3",
	# "rich>=13.6.0",
	# ]
	# ///
	"""
	Validate and auto-update security documentation against config.yml.

	Checks performed:
	1. Every ``[section] option`` reference in the security RST files corresponds to an
	actual option in config.yml or provider.yaml.
	2. Default values quoted in the docs match the defaults in config.yml.
	3. Auto-updates the sensitive-variable tables in security_model.rst between
	AUTOGENERATED markers to stay in sync with config.yml and provider.yaml.
	"""

	from __future__ import annotations

	import re
	import sys
	from pathlib import Path

	import yaml
	from rich.console import Console

	sys.path.insert(0, str(Path(__file__).parent.resolve()))

	from common_prek_utils import AIRFLOW_ROOT_PATH

	console = Console(color_system="standard", width=200)

	CONFIG_YML = AIRFLOW_ROOT_PATH / "airflow-core" / "src" / "airflow" / "config_templates" / "config.yml"
	PROVIDERS_ROOT = AIRFLOW_ROOT_PATH / "providers"
	SECURITY_MODEL_RST = AIRFLOW_ROOT_PATH / "airflow-core" / "docs" / "security" / "security_model.rst"

	SECURITY_DOCS = [
	AIRFLOW_ROOT_PATH / "airflow-core" / "docs" / "security" / "jwt_token_authentication.rst",
	SECURITY_MODEL_RST,
	]

	# Pattern to match ``[section] option_name`` references in RST
	SECTION_OPTION_RE = re.compile(r"``\[(\w+)\]\s+(\w+)``")

	# Pattern to match AIRFLOW__SECTION__OPTION env var references
	ENV_VAR_RE = re.compile(r"``(AIRFLOW__\w+)``")

	# Map section+option to the AIRFLOW__ env var form
	SECTION_OPT_TO_ENV = re.compile(r"AIRFLOW__([A-Z_]+)__([A-Z_]+)")

	# Markers for autogenerated sections
	CORE_START = " .. START AUTOGENERATED CORE SENSITIVE VARS"
	CORE_END = " .. END AUTOGENERATED CORE SENSITIVE VARS"
	PROVIDER_START = " .. START AUTOGENERATED PROVIDER SENSITIVE VARS"
	PROVIDER_END = " .. END AUTOGENERATED PROVIDER SENSITIVE VARS"

	# Which components need which core config sections/options.
	# Maps (section, option) -> list of component names.
	# This is the source of truth for the "Needed by" column.
	CORE_COMPONENT_MAP: dict[tuple[str, str], str] = {
	("api", "secret_key"): "API Server, Scheduler, Workers, Triggerer",
	("api_auth", "jwt_secret"): "API Server, Scheduler",
	("core", "asset_manager_kwargs"): "Dag File Processor",
	("core", "fernet_key"): "API Server, Scheduler, Workers, Dag File Processor, Triggerer",
	("database", "sql_alchemy_conn"): "API Server, Scheduler, Dag File Processor, Triggerer",
	("database", "sql_alchemy_conn_async"): "API Server, Scheduler, Dag File Processor, Triggerer",
	("database", "sql_alchemy_engine_args"): "API Server, Scheduler, Dag File Processor, Triggerer",
	("logging", "remote_task_handler_kwargs"): "Scheduler, Workers, Triggerer",
	("secrets", "backend_kwargs"): "Scheduler, Dag File Processor, Triggerer",
	("sentry", "sentry_dsn"): "Scheduler, Triggerer",
	("workers", "secrets_backend_kwargs"): "Workers",
	}

	# Human-readable descriptions for core sensitive vars
	CORE_DESCRIPTIONS: dict[tuple[str, str], str] = {
	("api", "secret_key"): "API secret key for log token signing",
	("api_auth", "jwt_secret"): "JWT signing key (symmetric mode)",
	("core", "asset_manager_kwargs"): "Asset manager credentials",
	("core", "fernet_key"): "Fernet encryption key for connections/variables at rest",
	("database", "sql_alchemy_conn"): "Metadata database connection string",
	("database", "sql_alchemy_conn_async"): "Async metadata database connection string",
	("database", "sql_alchemy_engine_args"): "SQLAlchemy engine parameters (may contain credentials)",
	("logging", "remote_task_handler_kwargs"): "Remote logging handler credentials",
	("secrets", "backend_kwargs"): "Secrets backend credentials (non-worker mode)",
	("sentry", "sentry_dsn"): "Sentry error reporting endpoint",
	("workers", "secrets_backend_kwargs"): "Worker-specific secrets backend credentials",
	}


	def option_to_env_var(section: str, option: str) -> str:
	"""Convert a config section+option to its AIRFLOW__ env var form."""
	return f"AIRFLOW__{section.upper()}__{option.upper()}"


	def load_core_config() -> dict:
	"""Load the core config.yml."""
	with open(CONFIG_YML) as f:
	return yaml.safe_load(f)


	def load_provider_configs() -> dict[str, dict]:
	"""Load provider.yaml files. Returns {provider_name: config_sections}."""
	result = {}
	for provider_yaml in sorted(PROVIDERS_ROOT.glob("*/provider.yaml")):
	with open(provider_yaml) as f:
	data = yaml.safe_load(f)
	if data and "config" in data:
	provider_name = provider_yaml.parent.name
	result[provider_name] = data["config"]
	return result


	def get_all_options(core_config: dict, provider_configs: dict[str, dict]) -> dict[tuple[str, str], dict]:
	"""Return a dict of (section, option) -> option_config for all config options."""
	result = {}
	for section_name, section_data in core_config.items():
	if not isinstance(section_data, dict) or "options" not in section_data:
	continue
	for option_name, option_config in section_data["options"].items():
	if isinstance(option_config, dict):
	result[(section_name, option_name)] = option_config

	for _provider_name, sections in provider_configs.items():
	for section_name, section_data in sections.items():
	if not isinstance(section_data, dict) or "options" not in section_data:
	continue
	for option_name, option_config in section_data["options"].items():
	if isinstance(option_config, dict):
	result[(section_name, option_name)] = option_config

	return result


	def get_core_sensitive_vars(core_config: dict) -> list[tuple[str, str]]:
	"""Return sorted list of (section, option) for core sensitive config options."""
	result = []
	for section_name, section_data in core_config.items():
	if not isinstance(section_data, dict) or "options" not in section_data:
	continue
	for option_name, option_config in section_data["options"].items():
	if isinstance(option_config, dict) and option_config.get("sensitive"):
	result.append((section_name, option_name))
	return sorted(result, key=lambda x: option_to_env_var(*x))


	def get_provider_sensitive_vars(
	provider_configs: dict[str, dict],
	) -> list[tuple[str, str, str]]:
	"""Return sorted list of (provider, section, option) for provider sensitive config options."""
	result = []
	for provider_name, sections in provider_configs.items():
	for section_name, section_data in sections.items():
	if not isinstance(section_data, dict) or "options" not in section_data:
	continue
	for option_name, option_config in section_data["options"].items():
	if isinstance(option_config, dict) and option_config.get("sensitive"):
	result.append((provider_name, section_name, option_name))
	return sorted(result, key=lambda x: option_to_env_var(x[1], x[2]))


	def generate_core_table(core_sensitive: list[tuple[str, str]]) -> list[str]:
	"""Generate RST list-table lines for core sensitive vars."""
	lines = [
	"",
	" Core Airflow sensitive configuration variables:",
	"",
	" .. list-table::",
	" :header-rows: 1",
	" :widths: 40 30 30",
	"",
	" * - Environment variable",
	" - Description",
	" - Needed by",
	]
	for section, option in core_sensitive:
	env_var = option_to_env_var(section, option)
	desc = CORE_DESCRIPTIONS.get((section, option), f"[{section}] {option}")
	needed_by = CORE_COMPONENT_MAP.get((section, option), "Review per deployment")
	lines.append(f" * - ``{env_var}``")
	lines.append(f" - {desc}")
	lines.append(f" - {needed_by}")

	# Check for unmapped vars and warn
	for section, option in core_sensitive:
	if (section, option) not in CORE_COMPONENT_MAP:
	console.print(
	f" [yellow]⚠[/] New core sensitive var [{section}] {option} — "
	f"add it to CORE_COMPONENT_MAP in check_security_doc_constants.py"
	)
	if (section, option) not in CORE_DESCRIPTIONS:
	console.print(
	f" [yellow]⚠[/] New core sensitive var [{section}] {option} — "
	f"add a description to CORE_DESCRIPTIONS in check_security_doc_constants.py"
	)

	return lines


	def generate_provider_table(provider_sensitive: list[tuple[str, str, str]]) -> list[str]:
	"""Generate RST list-table lines for provider sensitive vars."""
	lines = [
	"",
	" Provider-specific sensitive configuration variables:",
	"",
	" The following variables are defined by Airflow providers and should only be set on components where",
	" the corresponding provider functionality is needed. The decision of which components require these",
	" variables depends on the Deployment Manager's choices about which providers and features are",
	" enabled in each component.",
	"",
	" .. list-table::",
	" :header-rows: 1",
	" :widths: 40 30 30",
	"",
	" * - Environment variable",
	" - Provider",
	" - Description",
	]
	for provider, section, option in provider_sensitive:
	env_var = option_to_env_var(section, option)
	# Generate a reasonable description from the option name
	desc = option.replace("_", " ").capitalize()
	lines.append(f" * - ``{env_var}``")
	lines.append(f" - {provider}")
	lines.append(f" - {desc}")

	return lines


	def update_autogenerated_section(
	content: str, start_marker: str, end_marker: str, new_lines: list[str]
	) -> str:
	"""Replace content between markers with new content."""
	lines = content.splitlines()
	start_idx = None
	end_idx = None

	for i, line in enumerate(lines):
	if start_marker in line:
	start_idx = i
	elif end_marker in line:
	end_idx = i
	break

	if start_idx is None or end_idx is None:
	console.print(f" [red]✗[/] Could not find markers {start_marker!r} / {end_marker!r}")
	return content

	result = lines[: start_idx + 1] + new_lines + [""] + lines[end_idx:]
	return "\n".join(result) + "\n"


	def update_sensitive_var_tables(
	core_sensitive: list[tuple[str, str]],
	provider_sensitive: list[tuple[str, str, str]],
	) -> bool:
	"""Update the autogenerated tables in security_model.rst. Returns True if changed."""
	content = SECURITY_MODEL_RST.read_text()
	original = content

	core_lines = generate_core_table(core_sensitive)
	content = update_autogenerated_section(content, CORE_START, CORE_END, core_lines)

	provider_lines = generate_provider_table(provider_sensitive)
	content = update_autogenerated_section(content, PROVIDER_START, PROVIDER_END, provider_lines)

	if content != original:
	SECURITY_MODEL_RST.write_text(content)
	return True
	return False


	def check_option_references(doc_path: Path, all_options: dict[tuple[str, str], dict]) -> list[str]:
	"""Check that all [section] option references in the doc exist in config.yml."""
	errors = []
	content = doc_path.read_text()

	for line_num, line in enumerate(content.splitlines(), 1):
	for match in SECTION_OPTION_RE.finditer(line):
	section = match.group(1)
	option = match.group(2)
	if (section, option) not in all_options:
	section_exists = any(s == section for s, _ in all_options)
	if section_exists:
	errors.append(
	f"{doc_path.name}:{line_num}: Option ``[{section}] {option}`` not found in config.yml"
	)
	else:
	errors.append(
	f"{doc_path.name}:{line_num}: Section ``[{section}]`` not found in config.yml"
	)
	return errors


	def check_env_var_references(doc_path: Path, all_options: dict[tuple[str, str], dict]) -> list[str]:
	"""Check that AIRFLOW__X__Y env var references correspond to real config options."""
	errors = []
	content = doc_path.read_text()

	for line_num, line in enumerate(content.splitlines(), 1):
	# Skip lines inside autogenerated sections — those are managed by the update logic
	if "AUTOGENERATED" in line:
	continue
	for match in ENV_VAR_RE.finditer(line):
	env_var = match.group(1)
	m = SECTION_OPT_TO_ENV.match(env_var)
	if not m:
	continue
	section = m.group(1).lower()
	option = m.group(2).lower()
	if (section, option) not in all_options:
	section_exists = any(s == section for s, _ in all_options)
	if section_exists:
	errors.append(
	f"{doc_path.name}:{line_num}: Env var ``{env_var}`` references "
	f"option [{section}] {option} which is not in config.yml"
	)
	else:
	errors.append(
	f"{doc_path.name}:{line_num}: Env var ``{env_var}`` references "
	f"section [{section}] which is not in config.yml"
	)
	return errors


	def check_defaults_in_tables(doc_path: Path, all_options: dict[tuple[str, str], dict]) -> list[str]:
	"""Check default values in RST table rows match config.yml."""
	errors = []
	content = doc_path.read_text()
	lines = content.splitlines()

	i = 0
	while i < len(lines):
	line = lines[i]
	match = SECTION_OPTION_RE.search(line)
	if match and "* -" in line:
	section = match.group(1)
	option = match.group(2)
	j = i + 1
	while j < len(lines) and not lines[j].strip():
	j += 1
	if j < len(lines) and lines[j].strip().startswith("-"):
	value_line = lines[j].strip().lstrip("- ").strip()
	key = (section, option)
	if key in all_options:
	config_default = str(all_options[key].get("default", "~"))
	doc_value = value_line.split()[0] if value_line else ""
	doc_value = doc_value.strip("`")
	config_default_clean = config_default.strip('"').strip("'")
	if (
	doc_value
	and config_default_clean
	and config_default_clean not in ("~", "None", "none", "")
	and doc_value != config_default_clean
	and not doc_value.startswith("Auto")
	and not doc_value.startswith("None")
	and doc_value != "``GUESS``"
	):
	errors.append(
	f"{doc_path.name}:{j + 1}: Default for [{section}] {option} is "
	f"'{doc_value}' in docs but '{config_default_clean}' in config.yml"
	)
	i += 1

	return errors


	def main() -> int:
	core_config = load_core_config()
	provider_configs = load_provider_configs()
	all_options = get_all_options(core_config, provider_configs)

	# Step 1: Auto-update the sensitive var tables
	core_sensitive = get_core_sensitive_vars(core_config)
	provider_sensitive = get_provider_sensitive_vars(provider_configs)

	if update_sensitive_var_tables(core_sensitive, provider_sensitive):
	console.print(
	" [yellow]⚠[/] security_model.rst sensitive variable tables were out of date and have been updated."
	)
	console.print(" [yellow] Please review and commit the changes.[/]")

	# Step 2: Validate references (re-read after potential update)
	all_errors: list[str] = []

	for doc_path in SECURITY_DOCS:
	if not doc_path.exists():
	console.print(f" [yellow]⚠[/] {doc_path.name} not found, skipping")
	continue
	all_errors.extend(check_option_references(doc_path, all_options))
	all_errors.extend(check_env_var_references(doc_path, all_options))
	all_errors.extend(check_defaults_in_tables(doc_path, all_options))

	if all_errors:
	console.print()
	for error in all_errors:
	console.print(f" [red]✗[/] {error}")
	console.print()
	console.print(f"[red]Security doc constants check failed with {len(all_errors)} error(s).[/]")
	console.print(
	"[yellow]Fix the documentation to match config.yml, or update config.yml if the docs are correct.[/]"
	)
	return 1

	console.print("[green]Security doc constants check passed.[/]")
	return 0


	if __name__ == "__main__":
	sys.exit(main())