diff --git a/.gitignore b/.gitignore
index 9c0755c..3354a6a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,6 +14,7 @@ htmlcov/
.env
.devops_learn/
artifacts/security/
+artifacts/deployment/
# Terraform working-directory artifacts (see projects/api_platform/infra/terraform/).
# .terraform.lock.hcl is deliberately NOT listed here: it pins exact provider
@@ -25,6 +26,8 @@ artifacts/security/
**/.terraform/
*.tfstate
*.tfstate.*
+*.tfplan
+*.tfplan.json
crash.log
crash.*.log
override.tf
diff --git a/README.md b/README.md
index dfffd6c..0fa6d29 100644
--- a/README.md
+++ b/README.md
@@ -39,6 +39,13 @@ Project -> Analysis -> Requirements -> Questions -> Recommendations -> Architect
evidence collection, compares an explicit Git base against the proposed
state, applies Conftest policy, redacts secrets structurally, and prevents
blocked changes from progressing toward deployment eligibility.
+- **Azure release lifecycle implementation** (`devops-learn deploy
+ projects/api_platform --cloud azure --depth learning`) composes validation,
+ image scanning, saved Terraform plans, candidate-bound approvals, ACR push,
+ Container Apps deployment, Azure observation, and HTTP health verification.
+ It is real-only and stops at preflight when credentials or required tools are
+ absent. It has not been executed from this checkout because Azure CLI,
+ Docker, Trivy, and Conftest are unavailable here.
- **Competency evidence tracking**: records what you were exposed to, practiced,
or demonstrated, not fabricated certification.
@@ -67,6 +74,12 @@ devops-learn local projects/api_platform
# (plan requires Azure credentials; fails cleanly with an explanation without them)
devops-learn terraform
+# Run the complete real-only Azure lifecycle after `az login` and tool preflight.
+devops-learn deploy projects/api_platform --cloud azure --depth learning
+
+# Explicitly approve and verify cleanup when finished learning.
+devops-learn destroy projects/api_platform
+
# Check security prerequisites, then scan a proposed change against its base
devops-learn security doctor
devops-learn security scan projects/api_platform --base-ref origin/main
diff --git a/docs/devsecops.md b/docs/devsecops.md
index a74d546..401adfb 100644
--- a/docs/devsecops.md
+++ b/docs/devsecops.md
@@ -134,10 +134,11 @@ the bundled real FastAPI project, uploads the sanitized report, and never uses
Azure credentials or runs `terraform apply`. GitHub PR metadata supplies the
base SHA, not `HEAD~1`.
-The future real path is: Terraform fmt, validate, plan, plan JSON risk
-analysis, Trivy IaC/plan scan, normalized change-aware findings, Conftest
-policy, deployment eligibility, human plan approval, real apply, then Azure
-verification. Azure and apply are not implemented by this milestone.
+The real-only `deploy` path composes Terraform fmt, validate and saved plan,
+plan JSON risk analysis, Trivy filesystem/config/image evidence, normalized
+change-aware findings, Conftest policy, deployment eligibility, human approval,
+candidate-bound real apply, Azure observation, and health verification. It is
+not marked live-verified until an opt-in Azure run succeeds.
## Limitations
diff --git a/docs/index.html b/docs/index.html
index 9a406b1..b04f04c 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -5,7 +5,7 @@
-
+
@@ -596,10 +596,11 @@
One project. A full DevOps journey.
03 // Truthful status
-
Real local execution. Simulated cloud.
+
Real local execution. Real-only Azure lifecycle.
- The boundary is deliberate: local Python and Docker work is real, while Azure,
- Terraform apply, and Kubernetes remain simulated until the next milestone.
+ The boundary is deliberate: local work, saved Terraform plans, and the Azure
+ lifecycle implementation are real-only. Azure execution remains opt-in and is
+ never claimed as verified until a credentialed run succeeds.
@@ -618,7 +619,7 @@
Real local execution. Simulated cloud.
Sim
-
Azure provisioning, Terraform apply, kubectl, and cloud API calls are simulated.
+
Azure provisioning and Terraform apply require explicit approval and opt-in credentials; Kubernetes remains simulated.
@@ -663,11 +664,11 @@
Real Python + Docker
Real Terraform validation
-
Use the Terraform binary for validate/plan while Azure apply/destroy remains simulated.
+
Use saved Terraform plans with explicit approval before real-only apply and verified cleanup.
-
First real Azure path
-
Provision learner-owned Azure resources behind approval gates, then extend into real Kubernetes fundamentals.
+
Azure lifecycle acceptance run
+
Exercise the real, approval-controlled Container Apps path and independently verify cleanup.
diff --git a/docs/roadmap.md b/docs/roadmap.md
index 4578277..93863f1 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -41,15 +41,20 @@ declared on `RealTerraformTool` and stay simulated until Milestone 3. `plan`
requires Azure authentication and fails cleanly, with an explanation, on a
machine without it — `init`/`validate`/`fmt` need no credentials.
-## Milestone 3: first Azure vertical slice
-
-A small, real path: Python application -> Docker -> ACR -> Terraform -> minimal
-Azure infrastructure -> deployment -> health verification, behind the existing
-approval gate confirming a real, non-simulated action. Provision a resource
-group and container registry first. Introduce AKS/Kubernetes only after this
-first real cloud execution path is reliable — not because Kubernetes is
-technically harder, but because a working real deployment path should exist
-before adding cluster-level complexity on top of it.
+## Milestone 3: first Azure vertical slice (implemented, awaiting live verification)
+
+The real-only `deploy` workflow uses a two-stage path: bootstrap Resource Group,
+ACR, Log Analytics, Container Apps Environment, managed identity and AcrPull;
+then push a digest-pinned FastAPI image and apply the Container App plan.
+It persists saved plans and candidate evidence, requires explicit confirmation
+and tool approval, checks Azure independently, verifies `/health`, and offers
+an approved `destroy` flow with an Azure cleanup check. Container Apps is the
+right first fit for a small HTTP workload: it is managed and has much less
+operational overhead than AKS. AKS remains deliberately out of scope.
+
+This feature is not yet labelled live-verified: the current execution host has
+no Azure CLI, Docker, Trivy, or Conftest, so a real Azure deployment and cleanup
+remain an opt-in acceptance test.
## DevSecOps security control plane (done)
diff --git a/docs/safety.md b/docs/safety.md
index 51c59c3..892e785 100644
--- a/docs/safety.md
+++ b/docs/safety.md
@@ -78,9 +78,12 @@ resources, applies Terraform, or modifies remote state. It:
(`tools/terraform_tool.py`) — `devops-learn analyze --real-tools` and `devops-learn local` both
still use `SimulatedTerraformTool` for the `terraform` tool name; only `python`/`docker` become
real for those commands. `RealTerraformTool` declares only `fmt`/`init`/`validate`/`plan` —
-`apply_approved_plan`/`destroy_approved_environment` are not implemented on it at all, so
-invoking either raises `KeyError` rather than silently falling back to simulated behavior. Real
-apply/destroy stay simulated until Milestone 3 (`docs/roadmap.md`).
+`apply_approved_plan` and `destroy_approved_environment` are real-only Tool
+operations. Apply accepts only a saved plan in `.devops_learn/plans` and
+rechecks its digest, source identity, Terraform configuration digest, and
+candidate context before invoking Terraform. Destroy requires an explicit
+target environment and is followed by a read-only Azure resource-group check.
+Neither operation can fall back to simulation; normal CI does not call either.
Every subprocess call goes through `tools/_subprocess_safety.py::run_safely`, which sets an
explicit timeout and always redacts secret-shaped `KEY=value` text and truncates long output
diff --git a/docs/terraform-state.md b/docs/terraform-state.md
index d384451..4b73cdc 100644
--- a/docs/terraform-state.md
+++ b/docs/terraform-state.md
@@ -88,7 +88,6 @@ DESTRUCTIVE-adjacent operations in this platform's risk model (see
## Real vs. simulated in this platform
`devops-learn terraform` runs real `fmt`/`init`/`validate`/`plan` against
-`projects/api_platform/infra/terraform/` and never applies or destroys.
-`apply_approved_plan`/`destroy_approved_environment` -- and therefore any
-real state file this platform would ever write -- stay simulated until
-Milestone 3. See `docs/safety.md` and `docs/roadmap.md`.
+`projects/api_platform/infra/terraform/`. The unified `deploy` command adds
+real-only saved-plan apply and the `destroy` command adds real-only cleanup.
+Both remain unverified in this checkout until an opt-in Azure acceptance run.
diff --git a/projects/api_platform/infra/terraform/README.md b/projects/api_platform/infra/terraform/README.md
index d2b808f..a081afb 100644
--- a/projects/api_platform/infra/terraform/README.md
+++ b/projects/api_platform/infra/terraform/README.md
@@ -22,8 +22,11 @@ requires Azure authentication (`az login`, or `ARM_CLIENT_ID` /
variables) to build the provider client -- without it, `plan` fails cleanly
and the workflow explains why rather than faking a result.
-`apply`/`destroy` are not implemented against this directory yet; see
-`docs/roadmap.md` (Milestone 3) and `docs/terraform-state.md`.
+`devops-learn deploy` persists an approved plan before real apply; it never
+accepts an arbitrary `terraform apply` command. `devops-learn destroy` is
+separately destructive, requires explicit confirmation and tool approval, and
+checks Azure after Terraform returns. These real-only operations are excluded
+from CI and have not been exercised from this checkout without Azure credentials.
## Local state
diff --git a/projects/api_platform/infra/terraform/main.tf b/projects/api_platform/infra/terraform/main.tf
index 1ec84b1..df711d4 100644
--- a/projects/api_platform/infra/terraform/main.tf
+++ b/projects/api_platform/infra/terraform/main.tf
@@ -32,6 +32,19 @@ resource "azurerm_container_registry" "main" {
tags = local.tags
}
+resource "azurerm_user_assigned_identity" "container_app" {
+ name = "${var.project_name}-app-identity"
+ resource_group_name = azurerm_resource_group.main.name
+ location = azurerm_resource_group.main.location
+ tags = local.tags
+}
+
+resource "azurerm_role_assignment" "container_app_acr_pull" {
+ scope = azurerm_container_registry.main.id
+ role_definition_name = "AcrPull"
+ principal_id = azurerm_user_assigned_identity.container_app.principal_id
+}
+
resource "azurerm_log_analytics_workspace" "main" {
name = "${var.project_name}-logs"
resource_group_name = azurerm_resource_group.main.name
@@ -40,3 +53,55 @@ resource "azurerm_log_analytics_workspace" "main" {
retention_in_days = 30
tags = local.tags
}
+
+resource "azurerm_container_app_environment" "main" {
+ name = "${var.project_name}-env"
+ resource_group_name = azurerm_resource_group.main.name
+ location = azurerm_resource_group.main.location
+ log_analytics_workspace_id = azurerm_log_analytics_workspace.main.id
+ tags = local.tags
+}
+
+resource "azurerm_container_app" "api" {
+ count = var.deploy_application ? 1 : 0
+ name = "${var.project_name}-api"
+ resource_group_name = azurerm_resource_group.main.name
+ container_app_environment_id = azurerm_container_app_environment.main.id
+ revision_mode = "Single"
+
+ identity {
+ type = "UserAssigned"
+ identity_ids = [azurerm_user_assigned_identity.container_app.id]
+ }
+
+ registry {
+ server = azurerm_container_registry.main.login_server
+ identity = azurerm_user_assigned_identity.container_app.id
+ }
+
+ ingress {
+ external_enabled = true
+ target_port = 8000
+
+ traffic_weight {
+ latest_revision = true
+ percentage = 100
+ }
+ }
+
+ template {
+ min_replicas = 0
+ max_replicas = 1
+
+ container {
+ name = "api"
+ image = var.app_image
+ cpu = 0.25
+ memory = "0.5Gi"
+ }
+ }
+
+ tags = local.tags
+
+ depends_on = [azurerm_role_assignment.container_app_acr_pull]
+}
diff --git a/projects/api_platform/infra/terraform/outputs.tf b/projects/api_platform/infra/terraform/outputs.tf
index ea34d4c..6d6459c 100644
--- a/projects/api_platform/infra/terraform/outputs.tf
+++ b/projects/api_platform/infra/terraform/outputs.tf
@@ -5,3 +5,15 @@ output "resource_group_name" {
output "container_registry_login_server" {
value = azurerm_container_registry.main.login_server
}
+
+output "container_app_environment_name" {
+ value = azurerm_container_app_environment.main.name
+}
+
+output "container_app_name" {
+ value = try(azurerm_container_app.api[0].name, null)
+}
+
+output "container_app_endpoint" {
+ value = try("https://${azurerm_container_app.api[0].ingress[0].fqdn}", null)
+}
diff --git a/projects/api_platform/infra/terraform/variables.tf b/projects/api_platform/infra/terraform/variables.tf
index 93feebe..f69bb34 100644
--- a/projects/api_platform/infra/terraform/variables.tf
+++ b/projects/api_platform/infra/terraform/variables.tf
@@ -9,3 +9,20 @@ variable "location" {
type = string
default = "eastus"
}
+
+variable "deploy_application" {
+ description = "Set after the ACR image has been pushed. Bootstrap stays image-independent."
+ type = bool
+ default = false
+}
+
+variable "app_image" {
+ description = "Immutable ACR image reference, including @sha256 digest, for the FastAPI app."
+ type = string
+ default = ""
+
+ validation {
+ condition = !var.deploy_application || can(regex("@sha256:[0-9a-f]{64}$", var.app_image))
+ error_message = "Application deployment requires an immutable image digest reference."
+ }
+}
diff --git a/pyproject.toml b/pyproject.toml
index a7259a2..d352f88 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,8 +4,8 @@ build-backend = "hatchling.build"
[project]
name = "devops-learn"
-version = "0.1.0"
-description = "AI mentored, learn by doing DevOps and platform engineering CLI"
+version = "0.2.0"
+description = "AI-assisted DevOps learning CLI with a security-gated Azure deployment lifecycle"
readme = "README.md"
requires-python = ">=3.11"
license = { text = "MIT" }
diff --git a/src/devops_learn/cli/commands/__init__.py b/src/devops_learn/cli/commands/__init__.py
index e69de29..0e632d7 100644
--- a/src/devops_learn/cli/commands/__init__.py
+++ b/src/devops_learn/cli/commands/__init__.py
@@ -0,0 +1,4 @@
+from devops_learn.cli.commands import deploy
+from devops_learn.cli.commands import destroy
+
+__all__ = ["deploy", "destroy"]
diff --git a/src/devops_learn/cli/commands/deploy.py b/src/devops_learn/cli/commands/deploy.py
new file mode 100644
index 0000000..db630d7
--- /dev/null
+++ b/src/devops_learn/cli/commands/deploy.py
@@ -0,0 +1,41 @@
+"""CLI entry point for the first real Azure deployment lifecycle."""
+
+from __future__ import annotations
+
+import argparse
+
+from devops_learn.bootstrap import Platform
+from devops_learn.cli.terminal_ui import TerminalUi
+from devops_learn.domain.enums import ExplanationDepth
+from devops_learn.workflows.deploy_flow import DeployOptions, run_deploy_flow
+
+
+def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
+ parser = subparsers.add_parser(
+ "deploy",
+ help="Run the security-gated, approval-controlled real Azure deployment lifecycle",
+ )
+ parser.add_argument("path", nargs="?", default="projects/api_platform")
+ parser.add_argument("--cloud", choices=("azure",), default="azure")
+ parser.add_argument(
+ "--depth", choices=("brief", "normal", "learning", "deep"), default="learning"
+ )
+ parser.add_argument("--location", default="eastus")
+ parser.add_argument("--environment", default="learning")
+ parser.add_argument("--base-ref", default="origin/main")
+ parser.set_defaults(handler=run)
+
+
+def run(args: argparse.Namespace, platform: Platform) -> None:
+ run_deploy_flow(
+ platform,
+ TerminalUi(),
+ DeployOptions(
+ project_root=args.path,
+ cloud=args.cloud,
+ depth=ExplanationDepth[args.depth.upper()],
+ location=args.location,
+ environment=args.environment,
+ base_ref=args.base_ref,
+ ),
+ )
diff --git a/src/devops_learn/cli/commands/destroy.py b/src/devops_learn/cli/commands/destroy.py
new file mode 100644
index 0000000..be24466
--- /dev/null
+++ b/src/devops_learn/cli/commands/destroy.py
@@ -0,0 +1,35 @@
+"""CLI entry point for explicitly approved Azure learning-environment cleanup."""
+
+from __future__ import annotations
+
+import argparse
+
+from devops_learn.bootstrap import Platform
+from devops_learn.cli.terminal_ui import TerminalUi
+from devops_learn.domain.enums import ExplanationDepth
+from devops_learn.workflows.deploy_flow import DeployOptions, run_cleanup_flow
+
+
+def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
+ parser = subparsers.add_parser(
+ "destroy", help="Destroy and independently verify an Azure lab cleanup"
+ )
+ parser.add_argument("path", nargs="?", default="projects/api_platform")
+ parser.add_argument("--location", default="eastus")
+ parser.add_argument("--environment", default="learning")
+ parser.set_defaults(handler=run)
+
+
+def run(args: argparse.Namespace, platform: Platform) -> None:
+ success = run_cleanup_flow(
+ platform,
+ TerminalUi(),
+ DeployOptions(
+ project_root=args.path,
+ location=args.location,
+ environment=args.environment,
+ depth=ExplanationDepth.LEARNING,
+ ),
+ )
+ if not success:
+ raise SystemExit(2)
diff --git a/src/devops_learn/cli/main.py b/src/devops_learn/cli/main.py
index d80f9c0..03ca1af 100644
--- a/src/devops_learn/cli/main.py
+++ b/src/devops_learn/cli/main.py
@@ -9,6 +9,8 @@
from devops_learn.bootstrap import build_platform
from devops_learn.cli.commands import (
analyze,
+ deploy,
+ destroy,
explain,
history,
init,
@@ -32,8 +34,21 @@
from devops_learn.tools.cloud_tool import SimulatedCloudTool
from devops_learn.tools.policy_tool import PolicyTool
from devops_learn.tools.security_scanner_tool import SecurityScannerTool
+from devops_learn.tools.azure_tool import AzureCliTool
-_COMMAND_MODULES = (analyze, review, history, explain, profile, init, local, terraform, security)
+_COMMAND_MODULES = (
+ analyze,
+ review,
+ history,
+ explain,
+ profile,
+ init,
+ local,
+ terraform,
+ security,
+ deploy,
+ destroy,
+)
def build_parser() -> argparse.ArgumentParser:
@@ -95,6 +110,19 @@ def _tools_for_args(args: argparse.Namespace) -> dict[str, Tool] | None:
"cloud": SimulatedCloudTool(),
"validation": SimulatedValidationTool(),
}
+ if args.command in {"deploy", "destroy"}:
+ return {
+ "python": RealPythonTool(),
+ "git": SimulatedGitTool(),
+ "docker": RealDockerTool(),
+ "terraform": RealTerraformTool(),
+ "kubernetes": SimulatedKubernetesTool(),
+ "cloud": SimulatedCloudTool(),
+ "validation": SimulatedValidationTool(),
+ "security_scanner": SecurityScannerTool(),
+ "security_policy": PolicyTool(),
+ "azure": AzureCliTool(),
+ }
if getattr(args, "real_tools", False) or args.command == "local":
return {
"python": RealPythonTool(),
diff --git a/src/devops_learn/deployment/__init__.py b/src/devops_learn/deployment/__init__.py
new file mode 100644
index 0000000..b636c3f
--- /dev/null
+++ b/src/devops_learn/deployment/__init__.py
@@ -0,0 +1,5 @@
+"""Release candidate identity and real Azure deployment workflow."""
+
+from devops_learn.deployment.candidate import DeploymentCandidate
+
+__all__ = ["DeploymentCandidate"]
diff --git a/src/devops_learn/deployment/candidate.py b/src/devops_learn/deployment/candidate.py
new file mode 100644
index 0000000..da090ed
--- /dev/null
+++ b/src/devops_learn/deployment/candidate.py
@@ -0,0 +1,95 @@
+"""Immutable, deterministic evidence binding for a deployment.
+
+The candidate's identity deliberately excludes its creation timestamp. The
+same source, image, plan, and security evidence therefore produce the same
+identity, while the timestamp remains useful audit context. Approval records
+must name this identity; changing any material evidence creates a new one.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from hashlib import sha256
+import json
+from pathlib import Path
+
+
+def sha256_file(path: Path) -> str:
+ """Return a stable digest without ever rendering the file's content."""
+ digest = sha256()
+ with path.open("rb") as artifact:
+ for block in iter(lambda: artifact.read(1024 * 1024), b""):
+ digest.update(block)
+ return digest.hexdigest()
+
+
+@dataclass(frozen=True)
+class DeploymentCandidate:
+ """Exactly the deployable evidence a human is authorizing."""
+
+ source_revision: str
+ project_path: str
+ cloud: str
+ environment: str
+ terraform_config_digest: str
+ terraform_plan_path: str | None = None
+ terraform_plan_digest: str | None = None
+ terraform_plan_risk: str | None = None
+ security_report_path: str | None = None
+ security_report_digest: str | None = None
+ security_decision: str | None = None
+ image_reference: str | None = None
+ image_digest: str | None = None
+ deployment_eligibility: str | None = None
+ human_approvals: tuple[str, ...] = ()
+ created_at: datetime | None = None
+
+ def __post_init__(self) -> None:
+ if self.terraform_plan_digest and not self.terraform_plan_path:
+ raise ValueError("A Terraform plan digest requires a saved plan path.")
+ if self.security_report_digest and not self.security_report_path:
+ raise ValueError("A security report digest requires its report path.")
+ if self.image_digest and not self.image_reference:
+ raise ValueError("An image digest requires an image reference.")
+
+ @property
+ def identity(self) -> str:
+ """Stable SHA-256 identity of the material, approval-relevant state."""
+ encoded = json.dumps(self.material(), sort_keys=True, separators=(",", ":")).encode()
+ return sha256(encoded).hexdigest()
+
+ @property
+ def context_identity(self) -> str:
+ """Pre-plan binding for a saved plan; excludes only its own digest."""
+ material = self.material()
+ material["terraform_plan_digest"] = None
+ material["terraform_plan_risk"] = None
+ encoded = json.dumps(material, sort_keys=True, separators=(",", ":")).encode()
+ return sha256(encoded).hexdigest()
+
+ def material(self) -> dict[str, str | None]:
+ return {
+ "source_revision": self.source_revision,
+ "project_path": self.project_path,
+ "cloud": self.cloud,
+ "environment": self.environment,
+ "terraform_config_digest": self.terraform_config_digest,
+ "terraform_plan_digest": self.terraform_plan_digest,
+ "terraform_plan_risk": self.terraform_plan_risk,
+ "security_report_digest": self.security_report_digest,
+ "security_decision": self.security_decision,
+ "image_reference": self.image_reference,
+ "image_digest": self.image_digest,
+ }
+
+ def is_current(self) -> bool:
+ """Check that persisted evidence still matches the candidate."""
+ return all(
+ (
+ not self.terraform_plan_path
+ or self.terraform_plan_digest == sha256_file(Path(self.terraform_plan_path)),
+ not self.security_report_path
+ or self.security_report_digest == sha256_file(Path(self.security_report_path)),
+ )
+ )
diff --git a/src/devops_learn/deployment/evidence.py b/src/devops_learn/deployment/evidence.py
new file mode 100644
index 0000000..57b3f4b
--- /dev/null
+++ b/src/devops_learn/deployment/evidence.py
@@ -0,0 +1,31 @@
+"""Sanitized, verifiable lifecycle evidence reports."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
+
+from devops_learn.deployment.candidate import DeploymentCandidate
+from devops_learn.security.redaction import redact_data
+
+
+def write_evidence_report(
+ path: Path,
+ candidate: DeploymentCandidate,
+ *,
+ status: str,
+ stages: dict[str, str],
+ observed_azure: dict[str, Any] | None = None,
+) -> None:
+ """Persist facts only; raw tool output and secrets are intentionally absent."""
+ path.parent.mkdir(parents=True, exist_ok=True)
+ document = {
+ "candidate_identity": candidate.identity,
+ "candidate": asdict(candidate),
+ "status": status,
+ "stages": stages,
+ "observed_azure": observed_azure or {},
+ }
+ path.write_text(json.dumps(redact_data(document), indent=2, sort_keys=True, default=str) + "\n")
diff --git a/src/devops_learn/tools/azure_tool.py b/src/devops_learn/tools/azure_tool.py
new file mode 100644
index 0000000..b979d5c
--- /dev/null
+++ b/src/devops_learn/tools/azure_tool.py
@@ -0,0 +1,278 @@
+"""A deliberately narrow, real Azure CLI boundary.
+
+This tool is read-only except for ``acr_login``. It never accepts arbitrary
+Azure CLI arguments, never prints credentials, and makes the declaration vs.
+observation distinction explicit for the deployment workflow.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+from typing import Any, Mapping
+
+from devops_learn.tools import _subprocess_safety
+from devops_learn.tools.approval import ApprovalRecord
+from devops_learn.tools.base import RiskLevel, Tool, ToolOperationSpec, ToolResult
+
+_OPERATIONS = (
+ ToolOperationSpec("preflight", RiskLevel.SAFE, False, False, False),
+ ToolOperationSpec("verify_environment", RiskLevel.SAFE, False, False, False),
+ ToolOperationSpec("verify_cleanup", RiskLevel.SAFE, False, False, False),
+ ToolOperationSpec("container_app_evidence", RiskLevel.SAFE, False, False, False),
+ ToolOperationSpec("acr_login", RiskLevel.LOW, False, False, False),
+)
+
+
+def _az_command() -> str:
+ command = shutil.which("az")
+ if command is None:
+ raise FileNotFoundError("Azure CLI not found. Install it, then run `az login`.")
+ return command
+
+
+def _json(command: list[str], *, timeout: int = 45) -> tuple[bool, dict[str, Any], str]:
+ result = _subprocess_safety.run_safely(command, cwd=None, timeout=timeout)
+ if result.returncode != 0:
+ return False, {}, result.stderr or result.stdout
+ try:
+ parsed = json.loads(result.stdout)
+ except json.JSONDecodeError:
+ return False, {}, "Azure CLI returned malformed JSON."
+ return isinstance(parsed, dict), parsed if isinstance(parsed, dict) else {}, ""
+
+
+class AzureCliTool(Tool):
+ @property
+ def name(self) -> str:
+ return "azure"
+
+ @property
+ def operations(self) -> tuple[ToolOperationSpec, ...]:
+ return _OPERATIONS
+
+ def execute(
+ self,
+ operation: str,
+ params: Mapping[str, Any],
+ *,
+ dry_run: bool,
+ approval: ApprovalRecord | None,
+ ) -> ToolResult:
+ spec = self.spec_for(operation)
+ if dry_run:
+ return ToolResult(
+ True, f"Would run Azure {operation} (dry run)", {}, spec.risk_level, True, approval
+ )
+ try:
+ az = _az_command()
+ if operation == "preflight":
+ return self._preflight(az, params, spec, approval)
+ if operation == "acr_login":
+ registry = str(params.get("registry_name", ""))
+ if not registry:
+ raise ValueError("ACR login requires a registry name.")
+ result = _subprocess_safety.run_safely(
+ [az, "acr", "login", "--name", registry, "--only-show-errors"],
+ cwd=None,
+ timeout=90,
+ )
+ return ToolResult(
+ result.returncode == 0,
+ "(real) authenticated Docker to ACR"
+ if result.returncode == 0
+ else "(real, failed) ACR login failed",
+ {"returncode": result.returncode, "stderr": result.stderr},
+ spec.risk_level,
+ False,
+ approval,
+ )
+ resource_group = str(params.get("resource_group", ""))
+ if not resource_group:
+ raise ValueError("Azure verification requires a resource group.")
+ if operation == "verify_cleanup":
+ result = _subprocess_safety.run_safely(
+ [az, "group", "exists", "--name", resource_group], cwd=None, timeout=45
+ )
+ removed = result.returncode == 0 and result.stdout.strip().lower() == "false"
+ return ToolResult(
+ removed,
+ "(real) Azure resource group is gone"
+ if removed
+ else "(real, failed) Azure resource group still exists",
+ {"resource_group": resource_group, "exists": result.stdout.strip()},
+ spec.risk_level,
+ False,
+ approval,
+ )
+ if operation == "container_app_evidence":
+ app = str(params.get("container_app_name", ""))
+ command = [
+ az,
+ "containerapp",
+ "show",
+ "--name",
+ app,
+ "--resource-group",
+ resource_group,
+ "--output",
+ "json",
+ ]
+ ok, payload, error = _json(command)
+ safe = _container_app_view(payload)
+ return ToolResult(
+ ok,
+ "(real) collected Container App evidence"
+ if ok
+ else "(real, failed) could not collect Container App evidence",
+ safe if ok else {"error": error},
+ spec.risk_level,
+ False,
+ approval,
+ )
+ return self._verify_environment(az, params, spec, approval)
+ except (FileNotFoundError, OSError, ValueError) as exc:
+ return ToolResult(
+ False,
+ f"(real, failed) {exc}",
+ {"error": str(exc)},
+ spec.risk_level,
+ False,
+ approval,
+ )
+
+ def _preflight(
+ self,
+ az: str,
+ params: Mapping[str, Any],
+ spec: ToolOperationSpec,
+ approval: ApprovalRecord | None,
+ ) -> ToolResult:
+ ok, account, error = _json([az, "account", "show", "--output", "json"])
+ if not ok:
+ return ToolResult(
+ False,
+ "(real, failed) Azure authentication unavailable; run `az login`.",
+ {"error": error},
+ spec.risk_level,
+ False,
+ approval,
+ )
+ details = {
+ "subscription": str(account.get("name", "unknown")),
+ "subscription_id": str(account.get("id", "unknown")),
+ "tenant": str(account.get("tenantId", "unavailable")),
+ "region": str(params.get("region", "eastus")),
+ "environment": str(params.get("environment", "learning")),
+ }
+ return ToolResult(
+ True,
+ "(real) Azure authentication preflight passed",
+ details,
+ spec.risk_level,
+ False,
+ approval,
+ )
+
+ def _verify_environment(
+ self,
+ az: str,
+ params: Mapping[str, Any],
+ spec: ToolOperationSpec,
+ approval: ApprovalRecord | None,
+ ) -> ToolResult:
+ group = str(params["resource_group"])
+ commands = {
+ "resource_group": [az, "group", "show", "--name", group, "--output", "json"],
+ "acr": [
+ az,
+ "acr",
+ "show",
+ "--name",
+ str(params["acr_name"]),
+ "--resource-group",
+ group,
+ "--output",
+ "json",
+ ],
+ "environment": [
+ az,
+ "containerapp",
+ "env",
+ "show",
+ "--name",
+ str(params["container_environment_name"]),
+ "--resource-group",
+ group,
+ "--output",
+ "json",
+ ],
+ "container_app": [
+ az,
+ "containerapp",
+ "show",
+ "--name",
+ str(params["container_app_name"]),
+ "--resource-group",
+ group,
+ "--output",
+ "json",
+ ],
+ }
+ observed: dict[str, Any] = {}
+ for name, command in commands.items():
+ ok, payload, error = _json(command)
+ if not ok:
+ return ToolResult(
+ False,
+ f"(real, failed) Azure did not verify {name}",
+ {"error": error, "observed": observed},
+ spec.risk_level,
+ False,
+ approval,
+ )
+ observed[name] = (
+ _container_app_view(payload) if name == "container_app" else _resource_view(payload)
+ )
+ expected_region = str(params.get("expected_region", "")).lower()
+ actual_region = str(observed["resource_group"].get("location", "")).lower()
+ expected_tags = params.get("expected_tags", {})
+ actual_tags = observed["resource_group"].get("tags", {})
+ valid = (not expected_region or expected_region == actual_region) and all(
+ actual_tags.get(k) == v for k, v in expected_tags.items()
+ )
+ return ToolResult(
+ valid,
+ "(real) DECLARED BY TERRAFORM matches OBSERVED IN AZURE"
+ if valid
+ else "(real, failed) Azure observation does not match declared region or tags",
+ observed,
+ spec.risk_level,
+ False,
+ approval,
+ )
+
+
+def _resource_view(payload: dict[str, Any]) -> dict[str, Any]:
+ return {
+ "id": payload.get("id"),
+ "name": payload.get("name"),
+ "location": payload.get("location"),
+ "tags": payload.get("tags", {}),
+ }
+
+
+def _container_app_view(payload: dict[str, Any]) -> dict[str, Any]:
+ properties = payload.get("properties", {})
+ template = properties.get("template", {})
+ containers = template.get("containers", [])
+ ingress = properties.get("configuration", {}).get("ingress", {})
+ return {
+ "name": payload.get("name"),
+ "location": payload.get("location"),
+ "tags": payload.get("tags", {}),
+ "image": containers[0].get("image") if containers else None,
+ "endpoint": ingress.get("fqdn"),
+ "revision": properties.get("latestRevisionName"),
+ "running_status": properties.get("runningStatus"),
+ }
diff --git a/src/devops_learn/tools/docker_tool.py b/src/devops_learn/tools/docker_tool.py
index f9a1706..726c987 100644
--- a/src/devops_learn/tools/docker_tool.py
+++ b/src/devops_learn/tools/docker_tool.py
@@ -50,15 +50,15 @@
requires_approval=True,
is_destructive=True,
),
+ ToolOperationSpec("inspect_digest", RiskLevel.SAFE, False, False, False),
+ ToolOperationSpec("push", RiskLevel.LOW, False, False, False),
)
def _docker_command() -> str:
command = shutil.which("docker")
if command is None:
- raise FileNotFoundError(
- "Docker CLI not found. Install Docker or use simulation mode."
- )
+ raise FileNotFoundError("Docker CLI not found. Install Docker or use simulation mode.")
return command
@@ -104,6 +104,14 @@ def execute(
name = params.get("container", "sim-container-01")
summary = f"Stopped container {name} (simulated)"
details = {"container": name}
+ elif operation == "inspect_digest":
+ image = str(params.get("image", "api-platform:dev"))
+ summary = f"Image digest for {image} (simulated)"
+ details = {"image": image, "digest": "sha256:simulated"}
+ elif operation == "push":
+ image = str(params.get("image", "api-platform:dev"))
+ summary = f"Pushed {image} (simulated)"
+ details = {"image": image}
else: # remove_image
image = params.get("image", "api-platform:dev")
summary = f"Removed image {image} (simulated)"
@@ -157,9 +165,7 @@ def execute(
context = params.get("context", ".")
tag = params.get("tag", "api-platform:dev")
file_arg = params.get("dockerfile", "Dockerfile")
- completed = _run(
- [docker, "build", "-t", tag, "-f", file_arg, context]
- )
+ completed = _run([docker, "build", "-t", tag, "-f", file_arg, context])
success = completed.returncode == 0
if completed.stdout:
summary = completed.stdout.strip().splitlines()[-1]
@@ -206,6 +212,36 @@ def execute(
"returncode": completed.returncode,
"stderr": completed.stderr,
}
+ elif operation == "inspect_digest":
+ image = str(params.get("image", "api-platform:dev"))
+ completed = _run(
+ [docker, "image", "inspect", "--format", "{{index .RepoDigests 0}}", image]
+ )
+ digest = completed.stdout.strip()
+ success = completed.returncode == 0 and "@sha256:" in digest
+ summary = (
+ digest
+ if success
+ else "image digest unavailable; push the image to a registry first"
+ )
+ details = {
+ "image": image,
+ "digest": digest,
+ "returncode": completed.returncode,
+ "stderr": completed.stderr,
+ }
+ elif operation == "push":
+ image = str(params.get("image", ""))
+ if not image:
+ raise ValueError("Docker push requires an explicit image reference.")
+ completed = _run([docker, "push", image])
+ success = completed.returncode == 0
+ summary = f"Pushed {image}" if success else f"Failed to push {image}"
+ details = {
+ "image": image,
+ "returncode": completed.returncode,
+ "stderr": completed.stderr,
+ }
else: # remove_image
image = params.get("image", "api-platform:dev")
completed = _run([docker, "rmi", image])
@@ -216,7 +252,7 @@ def execute(
"returncode": completed.returncode,
"stderr": completed.stderr,
}
- except FileNotFoundError as exc:
+ except (FileNotFoundError, OSError, ValueError) as exc:
return ToolResult(
success=False,
summary=str(exc),
diff --git a/src/devops_learn/tools/terraform_tool.py b/src/devops_learn/tools/terraform_tool.py
index 4d65bea..1011d38 100644
--- a/src/devops_learn/tools/terraform_tool.py
+++ b/src/devops_learn/tools/terraform_tool.py
@@ -22,7 +22,8 @@
import json
import re
import shutil
-import tempfile
+from datetime import datetime, timezone
+from hashlib import sha256
from pathlib import Path
from typing import Any, Mapping
@@ -52,6 +53,7 @@
requires_approval=False,
is_destructive=False,
),
+ ToolOperationSpec("output", RiskLevel.SAFE, False, False, False),
ToolOperationSpec(
name="apply_approved_plan",
risk_level=RiskLevel.HIGH,
@@ -173,6 +175,21 @@ def execute(
requires_approval=False,
is_destructive=False,
),
+ ToolOperationSpec("output", RiskLevel.SAFE, False, False, False),
+ ToolOperationSpec(
+ name="apply_approved_plan",
+ risk_level=RiskLevel.HIGH,
+ supports_dry_run=False,
+ requires_approval=True,
+ is_destructive=False,
+ ),
+ ToolOperationSpec(
+ name="destroy_approved_environment",
+ risk_level=RiskLevel.DESTRUCTIVE,
+ supports_dry_run=False,
+ requires_approval=True,
+ is_destructive=True,
+ ),
)
@@ -185,6 +202,62 @@ def _terraform_command() -> str:
return command
+def _digest_file(path: Path) -> str:
+ digest = sha256()
+ with path.open("rb") as source:
+ for block in iter(lambda: source.read(1024 * 1024), b""):
+ digest.update(block)
+ return digest.hexdigest()
+
+
+def terraform_config_digest(working_dir: Path) -> str:
+ """Hash committed Terraform inputs, never state or generated provider files."""
+ digest = sha256()
+ paths = sorted(
+ [*working_dir.glob("*.tf"), working_dir / ".terraform.lock.hcl"],
+ key=lambda path: path.name,
+ )
+ for path in paths:
+ if not path.is_file():
+ continue
+ digest.update(path.name.encode())
+ digest.update(b"\0")
+ digest.update(path.read_bytes())
+ digest.update(b"\0")
+ return digest.hexdigest()
+
+
+def _controlled_plan_path(working_dir: Path, value: object | None) -> Path:
+ artifacts = (working_dir / ".devops_learn" / "plans").resolve()
+ candidate = Path(str(value)).resolve() if value is not None else artifacts / "terraform.tfplan"
+ if candidate.parent != artifacts or candidate.suffix != ".tfplan":
+ raise ValueError("Saved plans must be direct .tfplan files in .devops_learn/plans.")
+ artifacts.mkdir(parents=True, exist_ok=True)
+ return candidate
+
+
+def _plan_metadata_path(plan_path: Path) -> Path:
+ return plan_path.with_suffix(".tfplan.json")
+
+
+def _variable_args(params: Mapping[str, Any]) -> list[str]:
+ variables = params.get("variables", {})
+ if not isinstance(variables, Mapping):
+ raise ValueError("Terraform variables must be a mapping.")
+ allowed = {"deploy_application", "app_image"}
+ if set(variables) - allowed:
+ raise ValueError("Only deploy_application and app_image are accepted by this workflow.")
+ args: list[str] = []
+ for name in sorted(variables):
+ value = variables[name]
+ if name == "deploy_application" and not isinstance(value, bool):
+ raise ValueError("deploy_application must be boolean.")
+ if name == "app_image" and (not isinstance(value, str) or "@sha256:" not in value):
+ raise ValueError("app_image must be a digest-pinned image reference.")
+ args.extend(["-var", f"{name}={str(value).lower() if isinstance(value, bool) else value}"])
+ return args
+
+
def _details_from_plan_json(plan_json: Mapping[str, Any]) -> dict[str, Any]:
"""Extracts only {address, action} pairs from a real `terraform show -json`
plan -- never the resource_changes[].change.before/after attribute maps,
@@ -219,16 +292,12 @@ def _details_from_plan_json(plan_json: Mapping[str, Any]) -> dict[str, Any]:
class RealTerraformTool(Tool):
- """Runs real terraform fmt/init/validate/plan via subprocess.
-
- Deliberately does not declare apply_approved_plan or
- destroy_approved_environment: those stay simulated until Milestone 3, per
- docs/roadmap.md. A caller invoking either operation on this tool gets a
- KeyError from Tool.spec_for rather than a result that looks real but
- silently isn't -- apply/destroy remain real capabilities of the
- "terraform" tool name only via SimulatedTerraformTool, exactly how
- cli/main.py already selects real-vs-simulated per command today. See
- docs/adr/0004-controlled-tool-execution.md.
+ """Runs real Terraform only through bounded, auditable operations.
+
+ Saved plans live under the working directory's ``.devops_learn/plans``.
+ Apply reads a metadata sidecar created by this tool and rejects a plan,
+ source, config, candidate, or digest mismatch before invoking Terraform.
+ There is no real-to-simulated fallback.
"""
@property
@@ -260,7 +329,17 @@ def execute(
approval=approval,
)
- working_dir = params.get("path")
+ working_dir_value = params.get("path")
+ if not isinstance(working_dir_value, str) or not Path(working_dir_value).is_dir():
+ return ToolResult(
+ False,
+ "(real, failed) Terraform working directory is unavailable.",
+ {},
+ spec.risk_level,
+ dry_run,
+ approval,
+ )
+ working_dir = str(Path(working_dir_value).resolve())
timeout_override = params.get("timeout_seconds")
def _timeout(default: int) -> int:
@@ -275,8 +354,10 @@ def _timeout(default: int) -> int:
timeout=_timeout(30),
)
success = result.returncode == 0
- summary = "Configuration already formatted" if success else (
- "Configuration needs formatting"
+ summary = (
+ "Configuration already formatted"
+ if success
+ else ("Configuration needs formatting")
)
details: dict[str, Any] = {
"returncode": result.returncode,
@@ -327,59 +408,108 @@ def _timeout(default: int) -> int:
"diagnostic_count": len(diagnostics),
"stderr": result.stderr,
}
- else: # plan
- with tempfile.TemporaryDirectory() as tmp_dir:
- plan_path = str(Path(tmp_dir) / "tfplan.out")
- plan_result = _subprocess_safety.run_safely(
- [terraform, "plan", "-input=false", "-no-color", f"-out={plan_path}"],
- cwd=working_dir,
- timeout=_timeout(180),
+ elif operation == "plan":
+ plan_path = _controlled_plan_path(Path(working_dir), params.get("plan_path"))
+ plan_result = _subprocess_safety.run_safely(
+ [
+ terraform,
+ "plan",
+ "-input=false",
+ "-no-color",
+ *_variable_args(params),
+ f"-out={plan_path}",
+ ],
+ cwd=working_dir,
+ timeout=_timeout(180),
+ )
+ if plan_result.returncode != 0:
+ return ToolResult(
+ False,
+ "(real, failed) terraform plan failed",
+ {
+ "returncode": plan_result.returncode,
+ "stdout": plan_result.stdout,
+ "stderr": plan_result.stderr,
+ },
+ spec.risk_level,
+ dry_run,
+ approval,
)
- if plan_result.returncode != 0:
- return ToolResult(
- success=False,
- summary="(real, failed) terraform plan failed",
- details={
- "returncode": plan_result.returncode,
- "stdout": plan_result.stdout,
- "stderr": plan_result.stderr,
- },
- risk_level=spec.risk_level,
- was_dry_run=dry_run,
- approval=approval,
- )
- show_result = _subprocess_safety.run_safely(
- [terraform, "show", "-json", plan_path], cwd=working_dir, timeout=30
+ show_result = _subprocess_safety.run_safely(
+ [terraform, "show", "-json", str(plan_path)], cwd=working_dir, timeout=30
+ )
+ if show_result.returncode != 0:
+ return ToolResult(
+ False,
+ "(real, failed) terraform show -json failed",
+ {
+ "returncode": show_result.returncode,
+ "stderr": show_result.stderr,
+ },
+ spec.risk_level,
+ dry_run,
+ approval,
)
- if show_result.returncode != 0:
- return ToolResult(
- success=False,
- summary="(real, failed) terraform show -json failed",
- details={
- "returncode": show_result.returncode,
- "stderr": show_result.stderr,
- },
- risk_level=spec.risk_level,
- was_dry_run=dry_run,
- approval=approval,
- )
- try:
- plan_json = json.loads(show_result.stdout)
- except json.JSONDecodeError:
- return ToolResult(
- success=False,
- summary="(real, failed) Could not parse terraform show -json output",
- details={"error": "malformed JSON from terraform show -json"},
- risk_level=spec.risk_level,
- was_dry_run=dry_run,
- approval=approval,
- )
- details = _details_from_plan_json(plan_json)
- success = True
- summary = (
- f"Plan: {details['create']} to add, {details['change']} to change, "
- f"{details['replace']} to replace, {details['destroy']} to destroy."
+ try:
+ plan_json = json.loads(show_result.stdout)
+ except json.JSONDecodeError:
+ return ToolResult(
+ False,
+ "(real, failed) Could not parse terraform show -json output",
+ {"error": "malformed JSON from terraform show -json"},
+ spec.risk_level,
+ dry_run,
+ approval,
)
+ details = _details_from_plan_json(plan_json)
+ plan_digest = _digest_file(plan_path)
+ metadata = {
+ "candidate_context": params.get("candidate_context"),
+ "config_digest": terraform_config_digest(Path(working_dir)),
+ "created_at": datetime.now(timezone.utc).isoformat(),
+ "plan_digest": plan_digest,
+ "source_revision": params.get("source_revision"),
+ }
+ _plan_metadata_path(plan_path).write_text(
+ json.dumps(metadata, sort_keys=True) + "\n"
+ )
+ details.update(
+ {"plan_path": str(plan_path), "plan_digest": plan_digest, **metadata}
+ )
+ success = True
+ summary = (
+ f"Plan: {details['create']} to add, {details['change']} to change, "
+ f"{details['replace']} to replace, {details['destroy']} to destroy."
+ )
+ elif operation == "output":
+ result = _subprocess_safety.run_safely(
+ [terraform, "output", "-json"], cwd=working_dir, timeout=_timeout(30)
+ )
+ try:
+ raw_outputs = json.loads(result.stdout) if result.stdout else {}
+ except json.JSONDecodeError:
+ raw_outputs = {}
+ details = {
+ name: value.get("value")
+ for name, value in raw_outputs.items()
+ if name
+ in {
+ "resource_group_name",
+ "container_registry_login_server",
+ "container_app_environment_name",
+ "container_app_name",
+ "container_app_endpoint",
+ }
+ and isinstance(value, dict)
+ }
+ success = result.returncode == 0
+ summary = "Terraform outputs collected" if success else "terraform output failed"
+ elif operation == "apply_approved_plan":
+ return self._apply(
+ terraform, Path(working_dir), params, spec, approval, _timeout(300)
+ )
+ else:
+ return self._destroy(terraform, working_dir, params, spec, approval, _timeout(300))
except (OSError, FileNotFoundError) as exc:
return ToolResult(
success=False,
@@ -399,3 +529,100 @@ def _timeout(default: int) -> int:
was_dry_run=dry_run,
approval=approval,
)
+
+ def _apply(
+ self,
+ terraform: str,
+ working_dir: Path,
+ params: Mapping[str, Any],
+ spec: ToolOperationSpec,
+ approval: ApprovalRecord | None,
+ timeout: int,
+ ) -> ToolResult:
+ try:
+ plan_path = _controlled_plan_path(working_dir, params.get("plan_path"))
+ metadata = json.loads(_plan_metadata_path(plan_path).read_text())
+ expected = {
+ "candidate_context": params.get("candidate_context"),
+ "source_revision": params.get("source_revision"),
+ "config_digest": terraform_config_digest(working_dir),
+ "plan_digest": _digest_file(plan_path),
+ }
+ if any(not value or metadata.get(key) != value for key, value in expected.items()):
+ return ToolResult(
+ False,
+ "(real, refused) approved plan evidence is stale or changed",
+ {"reason": "candidate, source, configuration, or plan digest did not match"},
+ spec.risk_level,
+ False,
+ approval,
+ )
+ result = _subprocess_safety.run_safely(
+ [terraform, "apply", "-input=false", "-no-color", str(plan_path)],
+ cwd=str(working_dir),
+ timeout=timeout,
+ )
+ return ToolResult(
+ result.returncode == 0,
+ "(real) approved Terraform plan applied"
+ if result.returncode == 0
+ else "(real, failed) terraform apply failed",
+ {
+ "returncode": result.returncode,
+ "stdout": result.stdout,
+ "stderr": result.stderr,
+ "plan_digest": expected["plan_digest"],
+ },
+ spec.risk_level,
+ False,
+ approval,
+ )
+ except (OSError, ValueError, json.JSONDecodeError) as exc:
+ return ToolResult(
+ False,
+ f"(real, refused) {exc}",
+ {"error": str(exc)},
+ spec.risk_level,
+ False,
+ approval,
+ )
+
+ def _destroy(
+ self,
+ terraform: str,
+ working_dir: str,
+ params: Mapping[str, Any],
+ spec: ToolOperationSpec,
+ approval: ApprovalRecord | None,
+ timeout: int,
+ ) -> ToolResult:
+ if not params.get("resource_group") or not params.get("environment"):
+ return ToolResult(
+ False,
+ "(real, refused) destroy requires resource group and environment identity",
+ {},
+ spec.risk_level,
+ False,
+ approval,
+ )
+ result = _subprocess_safety.run_safely(
+ [terraform, "destroy", "-input=false", "-auto-approve", "-no-color"],
+ cwd=working_dir,
+ timeout=timeout,
+ )
+ return ToolResult(
+ result.returncode == 0,
+ "(real) Terraform environment destroyed"
+ if result.returncode == 0
+ else "(real, failed) terraform destroy failed",
+ {
+ "returncode": result.returncode,
+ "stdout": result.stdout,
+ "stderr": result.stderr,
+ "resource_group": str(params["resource_group"]),
+ "environment": str(params["environment"]),
+ },
+ spec.risk_level,
+ False,
+ approval,
+ )
diff --git a/src/devops_learn/workflows/deploy_flow.py b/src/devops_learn/workflows/deploy_flow.py
new file mode 100644
index 0000000..5c90bef
--- /dev/null
+++ b/src/devops_learn/workflows/deploy_flow.py
@@ -0,0 +1,484 @@
+"""One real, security-gated Azure Container Apps deployment lifecycle."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from datetime import datetime, timezone
+from hashlib import sha256
+from pathlib import Path
+import shutil
+from typing import Any, Mapping
+import urllib.error
+import urllib.request
+
+from devops_learn.bootstrap import Platform
+from devops_learn.deployment.candidate import DeploymentCandidate, sha256_file
+from devops_learn.deployment.evidence import write_evidence_report
+from devops_learn.domain.enums import (
+ AuditEventType,
+ CloudProviderKind,
+ CostPriority,
+ DeploymentEligibility,
+ EnvironmentKind,
+ ExecutionMode,
+ ExperienceState,
+ ExplanationDepth,
+ SecurityGateDecision,
+)
+from devops_learn.security.eligibility import evaluate_deployment_eligibility
+from devops_learn.tools import _subprocess_safety
+from devops_learn.tools.terraform_tool import terraform_config_digest
+from devops_learn.validation import terraform_plan_analysis
+from devops_learn.workflows.security_flow import SecurityOptions, run_security_scan
+from devops_learn.workflows.ui import Ui
+
+
+@dataclass(frozen=True)
+class DeployOptions:
+ project_root: str
+ cloud: str = "azure"
+ depth: ExplanationDepth = ExplanationDepth.LEARNING
+ location: str = "eastus"
+ environment: str = "learning"
+ base_ref: str = "origin/main"
+
+
+def run_deploy_flow(platform: Platform, ui: Ui, options: DeployOptions) -> None:
+ """Run bootstrap then application deployment, stopping at every failed gate."""
+ project = Path(options.project_root).resolve()
+ infra = project / "infra" / "terraform"
+ report_path = project / "artifacts" / "deployment" / "azure-deployment-evidence.json"
+ stages: dict[str, str] = {}
+ if options.cloud != "azure" or not infra.is_dir():
+ ui.present(
+ "This real lifecycle currently supports the bundled Azure Terraform project only."
+ )
+ return
+ session = platform.session_service.start(
+ project_root=str(project),
+ mode=ExecutionMode.COLLABORATIVE,
+ explanation_depth=options.depth,
+ cloud=CloudProviderKind.AZURE,
+ environment=EnvironmentKind.DEV,
+ cost_priority=CostPriority.LOWEST_COST,
+ simulation_mode=False,
+ )
+ assert session.id is not None
+ session_id = session.id
+ _audit(
+ platform,
+ session_id,
+ AuditEventType.SESSION_STARTED,
+ "Real Azure deployment lifecycle started",
+ )
+ try:
+ preflight = _preflight(platform, ui, options)
+ if not preflight:
+ stages["preflight"] = "NOT EXECUTED: prerequisites or Azure authentication unavailable"
+ return
+ stages["preflight"] = "REAL PASS"
+ if not ui.confirm(
+ "Azure resources will be created in the displayed subscription. Continue?",
+ default=False,
+ ):
+ stages["cloud_confirmation"] = "NOT EXECUTED: learner declined"
+ return
+ stages["cloud_confirmation"] = "APPROVED"
+
+ if not _validation(platform, project):
+ stages["application_validation"] = "REAL FAIL"
+ return
+ stages["application_validation"] = "REAL PASS"
+ local_tag = "api-platform:deployment-candidate"
+ if not platform.tool_service.invoke(
+ "docker",
+ "build",
+ {
+ "context": str(project),
+ "dockerfile": str(project / "Dockerfile"),
+ "tag": local_tag,
+ },
+ ).success:
+ stages["docker_build"] = "REAL FAIL"
+ return
+ stages["docker_build"] = "REAL PASS"
+
+ source = _source_identity(project)
+ config_digest = terraform_config_digest(infra)
+ bootstrap_security = _security(platform, ui, project, options, local_tag, "bootstrap")
+ if bootstrap_security is None:
+ stages["bootstrap_security"] = "REAL FAIL"
+ return
+ stages["bootstrap_security"] = bootstrap_security[1].value.upper()
+ bootstrap = DeploymentCandidate(
+ source_revision=source,
+ project_path=str(project),
+ cloud="azure",
+ environment=options.environment,
+ terraform_config_digest=config_digest,
+ security_report_path=str(bootstrap_security[0]),
+ security_report_digest=sha256_file(bootstrap_security[0]),
+ security_decision=bootstrap_security[1].value,
+ created_at=datetime.now(timezone.utc),
+ )
+ bootstrap_candidate = _plan_and_gate(
+ platform, ui, infra, bootstrap, {"deploy_application": False}
+ )
+ if bootstrap_candidate is None:
+ stages["bootstrap_plan_or_eligibility"] = "REAL FAIL OR NOT APPROVED"
+ return
+ stages["bootstrap_plan_or_eligibility"] = "REAL APPROVED"
+ if not _apply(platform, infra, bootstrap_candidate):
+ stages["bootstrap_apply"] = "REAL FAIL OR APPROVAL DENIED"
+ return
+ stages["bootstrap_apply"] = "REAL PASS"
+
+ outputs = platform.tool_service.invoke("terraform", "output", {"path": str(infra)})
+ if not outputs.success:
+ stages["bootstrap_outputs"] = "REAL FAIL"
+ return
+ registry_server = str(outputs.details.get("container_registry_login_server", ""))
+ if not registry_server:
+ stages["bootstrap_outputs"] = "REAL FAIL: ACR output missing"
+ return
+ stages["bootstrap_outputs"] = "REAL PASS"
+ registry_name = registry_server.split(".", maxsplit=1)[0]
+ image_tag = f"{registry_server}/api:{source.split(':', maxsplit=1)[0][:12]}"
+ if not platform.tool_service.invoke(
+ "azure", "acr_login", {"registry_name": registry_name}
+ ).success:
+ stages["acr_login"] = "REAL FAIL"
+ return
+ if (
+ not platform.tool_service.invoke(
+ "docker",
+ "build",
+ {
+ "context": str(project),
+ "dockerfile": str(project / "Dockerfile"),
+ "tag": image_tag,
+ },
+ ).success
+ or not platform.tool_service.invoke("docker", "push", {"image": image_tag}).success
+ ):
+ stages["acr_image_push"] = "REAL FAIL"
+ return
+ digest_result = platform.tool_service.invoke(
+ "docker", "inspect_digest", {"image": image_tag}
+ )
+ if not digest_result.success:
+ stages["image_digest"] = "REAL FAIL"
+ return
+ image_digest_ref = str(digest_result.details["digest"])
+ stages["acr_image_push"] = "REAL PASS"
+ stages["image_digest"] = "REAL PASS"
+
+ application_security = _security(
+ platform, ui, project, options, image_digest_ref, "application"
+ )
+ if application_security is None:
+ stages["application_security"] = "REAL FAIL"
+ return
+ stages["application_security"] = application_security[1].value.upper()
+ application = DeploymentCandidate(
+ source_revision=source,
+ project_path=str(project),
+ cloud="azure",
+ environment=options.environment,
+ terraform_config_digest=config_digest,
+ image_reference=image_digest_ref,
+ image_digest=image_digest_ref.rsplit("@", maxsplit=1)[-1],
+ security_report_path=str(application_security[0]),
+ security_report_digest=sha256_file(application_security[0]),
+ security_decision=application_security[1].value,
+ created_at=datetime.now(timezone.utc),
+ )
+ application_candidate = _plan_and_gate(
+ platform,
+ ui,
+ infra,
+ application,
+ {"deploy_application": True, "app_image": image_digest_ref},
+ )
+ if application_candidate is None:
+ stages["application_plan_or_eligibility"] = "REAL FAIL OR NOT APPROVED"
+ return
+ if not _apply(platform, infra, application_candidate):
+ stages["application_apply"] = "REAL FAIL OR APPROVAL DENIED"
+ return
+ stages["application_apply"] = "REAL PASS"
+ observed = _verify_azure(platform, outputs.details, options)
+ if observed is None:
+ stages["azure_verification"] = "REAL FAIL"
+ return
+ stages["azure_verification"] = "REAL PASS"
+ endpoint = str(observed.get("container_app", {}).get("endpoint", ""))
+ if not _health_check(endpoint):
+ stages["health_verification"] = "REAL FAIL: gather Container App evidence"
+ platform.tool_service.invoke(
+ "azure",
+ "container_app_evidence",
+ {
+ "resource_group": outputs.details["resource_group_name"],
+ "container_app_name": outputs.details["container_app_name"],
+ },
+ )
+ return
+ stages["health_verification"] = "REAL PASS"
+ _audit(
+ platform,
+ session_id,
+ AuditEventType.DEPLOYMENT_SUCCEEDED,
+ "Azure Container App health verified",
+ )
+ for concept in (
+ "Azure subscription",
+ "ACR",
+ "image digest",
+ "managed identity",
+ "AcrPull",
+ "Container Apps",
+ "Terraform state",
+ "security gate",
+ "health checks",
+ ):
+ platform.experience_tracker.record(
+ session_id, concept, "Practiced in real deployment", ExperienceState.PRACTICED
+ )
+ finally:
+ candidate = locals().get("application_candidate", locals().get("bootstrap_candidate", None))
+ if isinstance(candidate, DeploymentCandidate):
+ write_evidence_report(
+ report_path,
+ candidate,
+ status="completed"
+ if stages.get("health_verification") == "REAL PASS"
+ else "stopped",
+ stages=stages,
+ )
+ platform.session_service.complete(session)
+ ui.present(f"Sanitized lifecycle evidence: {report_path}")
+
+
+def _preflight(platform: Platform, ui: Ui, options: DeployOptions) -> bool:
+ unavailable = [
+ name
+ for name in ("terraform", "docker", "trivy", "conftest", "az")
+ if not shutil.which(name)
+ ]
+ if unavailable:
+ ui.present("AZURE PREFLIGHT FAILED: install " + ", ".join(unavailable) + ".")
+ return False
+ result = platform.tool_service.invoke(
+ "azure", "preflight", {"region": options.location, "environment": options.environment}
+ )
+ if result.success:
+ ui.present(
+ (
+ "AZURE TARGET\n\nSubscription: {subscription}\nTenant: {tenant}\n"
+ "Region: {region}\nEnvironment: {environment}"
+ ).format(**result.details)
+ )
+ else:
+ ui.present(result.summary)
+ return result.success
+
+
+def _validation(platform: Platform, project: Path) -> bool:
+ return all(
+ platform.tool_service.invoke("python", operation, params).success
+ for operation, params in (
+ ("run_tests", {"path": str(project)}),
+ ("run_lint", {"path": str(project), "paths": ["."]}),
+ )
+ )
+
+
+def _security(
+ platform: Platform, ui: Ui, project: Path, options: DeployOptions, image: str, stage: str
+) -> tuple[Path, SecurityGateDecision] | None:
+ path = project / "artifacts" / "deployment" / f"security-{stage}.json"
+ report = run_security_scan(
+ platform, ui, SecurityOptions(str(project), options.base_ref, image, str(path))
+ )
+ if report is None:
+ return None
+ return path, report.policy.decision
+
+
+def _plan_and_gate(
+ platform: Platform,
+ ui: Ui,
+ infra: Path,
+ candidate: DeploymentCandidate,
+ variables: dict[str, object],
+) -> DeploymentCandidate | None:
+ if not platform.tool_service.invoke("terraform", "fmt", {"path": str(infra)}).success:
+ return None
+ if not platform.tool_service.invoke("terraform", "init", {"path": str(infra)}).success:
+ return None
+ if not platform.tool_service.invoke("terraform", "validate", {"path": str(infra)}).success:
+ return None
+ planned = platform.tool_service.invoke(
+ "terraform",
+ "plan",
+ {
+ "path": str(infra),
+ "source_revision": candidate.source_revision,
+ "candidate_context": candidate.context_identity,
+ "variables": variables,
+ },
+ )
+ if not planned.success:
+ return None
+ risk = terraform_plan_analysis.analyze(planned.details).risk_level
+ completed = replace(
+ candidate,
+ terraform_plan_path=str(planned.details["plan_path"]),
+ terraform_plan_digest=str(planned.details["plan_digest"]),
+ terraform_plan_risk=risk.name,
+ )
+ security_approval = True
+ if candidate.security_decision == SecurityGateDecision.BLOCK.value:
+ ui.present("SECURITY BLOCK: deployment stopped.")
+ return None
+ if candidate.security_decision == SecurityGateDecision.REQUIRE_APPROVAL.value:
+ prompt = f"Security approval for candidate {completed.identity[:12]} required. Approve?"
+ security_approval = ui.confirm(prompt, default=False)
+ risk_prompt = (
+ f"Terraform plan risk is {risk.name}. "
+ f"Approve this exact candidate {completed.identity[:12]}?"
+ )
+ risk_approval = risk.name not in {"HIGH", "DESTRUCTIVE"} or ui.confirm(
+ risk_prompt, default=False
+ )
+ eligibility = evaluate_deployment_eligibility(
+ validation_passed=True,
+ gate=SecurityGateDecision(candidate.security_decision or "block"),
+ approval_granted=security_approval and risk_approval,
+ )
+ ui.present(f"DEPLOYMENT ELIGIBILITY: {eligibility.eligibility.value} ({eligibility.reason})")
+ if eligibility.eligibility is DeploymentEligibility.ELIGIBLE:
+ approvals: list[str] = []
+ if candidate.security_decision == SecurityGateDecision.REQUIRE_APPROVAL.value:
+ approvals.append("security")
+ if risk.name in {"HIGH", "DESTRUCTIVE"}:
+ approvals.append("terraform_plan_risk")
+ return replace(
+ completed,
+ deployment_eligibility=eligibility.eligibility.value,
+ human_approvals=tuple(approvals),
+ )
+ return None
+
+
+def _apply(platform: Platform, infra: Path, candidate: DeploymentCandidate) -> bool:
+ if not candidate.is_current() or candidate.terraform_config_digest != terraform_config_digest(
+ infra
+ ):
+ return False
+ result = platform.tool_service.invoke(
+ "terraform",
+ "apply_approved_plan",
+ {
+ "path": str(infra),
+ "plan_path": candidate.terraform_plan_path,
+ "candidate_identity": candidate.identity,
+ "candidate_context": candidate.context_identity,
+ "source_revision": candidate.source_revision,
+ },
+ )
+ return result.success
+
+
+def _verify_azure(
+ platform: Platform, outputs: Mapping[str, Any], options: DeployOptions
+) -> dict[str, Any] | None:
+ result = platform.tool_service.invoke(
+ "azure",
+ "verify_environment",
+ {
+ "resource_group": outputs["resource_group_name"],
+ "acr_name": str(outputs["container_registry_login_server"]).split(".")[0],
+ "container_environment_name": outputs["container_app_environment_name"],
+ "container_app_name": outputs["container_app_name"],
+ "expected_region": options.location,
+ "expected_tags": {"environment": options.environment, "managed-by": "terraform"},
+ },
+ )
+ return dict(result.details) if result.success else None
+
+
+def _health_check(endpoint: str) -> bool:
+ if not endpoint:
+ return False
+ try:
+ with urllib.request.urlopen(f"{endpoint.rstrip('/')}/health", timeout=20) as response:
+ return response.status == 200 and "healthy" in response.read().decode("utf-8").lower()
+ except (urllib.error.URLError, TimeoutError):
+ return False
+
+
+def _source_identity(project: Path) -> str:
+ digest = sha256()
+ for path in sorted(
+ [
+ project / "Dockerfile",
+ project / "requirements.txt",
+ *project.glob("app/**/*.py"),
+ *project.glob("infra/terraform/*.tf"),
+ project / "infra/terraform/.terraform.lock.hcl",
+ ]
+ ):
+ if path.is_file():
+ digest.update(str(path.relative_to(project)).encode())
+ digest.update(path.read_bytes())
+ revision = _subprocess_safety.run_safely(
+ ["git", "-C", str(project), "rev-parse", "HEAD"], cwd=None, timeout=15
+ )
+ git_sha = revision.stdout.strip() if revision.returncode == 0 else "no-git-sha"
+ return f"{git_sha}:workspace-{digest.hexdigest()}"
+
+
+def _audit(platform: Platform, session_id: int, event_type: AuditEventType, summary: str) -> None:
+ platform.audit_service.record(
+ session_id=session_id,
+ event_type=event_type,
+ occurred_at=datetime.now(timezone.utc),
+ summary=summary,
+ )
+
+
+def run_cleanup_flow(platform: Platform, ui: Ui, options: DeployOptions) -> bool:
+ """Destroy the named learning environment only after two human checkpoints."""
+ project = Path(options.project_root).resolve()
+ infra = project / "infra" / "terraform"
+ if not infra.is_dir() or not _preflight(platform, ui, options):
+ return False
+ resource_group = "api-platform-rg"
+ ui.present(
+ "DESTROY TARGET\n\n"
+ f"Resource group: {resource_group}\nEnvironment: {options.environment}\n"
+ f"State path: {infra / 'terraform.tfstate'}\n"
+ "Expected deletion: Container App, Container Apps Environment, managed identity, "
+ "AcrPull assignment, ACR, Log Analytics Workspace, Resource Group."
+ )
+ if not ui.confirm("Destroy this learning environment?", default=False):
+ return False
+ result = platform.tool_service.invoke(
+ "terraform",
+ "destroy_approved_environment",
+ {
+ "path": str(infra),
+ "resource_group": resource_group,
+ "environment": options.environment,
+ },
+ )
+ if not result.success:
+ ui.present(result.summary)
+ return False
+ verified = platform.tool_service.invoke(
+ "azure", "verify_cleanup", {"resource_group": resource_group}
+ )
+ ui.present(verified.summary)
+ return verified.success
diff --git a/tests/deployment/test_candidate.py b/tests/deployment/test_candidate.py
new file mode 100644
index 0000000..80da091
--- /dev/null
+++ b/tests/deployment/test_candidate.py
@@ -0,0 +1,39 @@
+from dataclasses import replace
+
+import pytest
+
+from devops_learn.deployment.candidate import DeploymentCandidate
+
+
+def _candidate() -> DeploymentCandidate:
+ return DeploymentCandidate(
+ source_revision="source-a",
+ project_path="projects/api_platform",
+ cloud="azure",
+ environment="learning",
+ terraform_config_digest="terraform-a",
+ image_reference="acr/api@sha256:abc",
+ image_digest="sha256:abc",
+ security_report_path="report.json",
+ security_report_digest="security-a",
+ security_decision="allow",
+ )
+
+
+def test_candidate_identity_is_deterministic_and_change_aware() -> None:
+ candidate = _candidate()
+ assert candidate.identity == _candidate().identity
+ assert candidate.identity != replace(candidate, source_revision="source-b").identity
+ assert candidate.identity != replace(candidate, security_report_digest="security-b").identity
+ assert candidate.identity != replace(candidate, image_digest="sha256:def").identity
+ assert (
+ candidate.context_identity
+ != replace(candidate, terraform_config_digest="terraform-b").context_identity
+ )
+
+
+def test_candidate_requires_evidence_paths_for_digests() -> None:
+ with pytest.raises(ValueError, match="saved plan path"):
+ replace(_candidate(), terraform_plan_digest="plan")
+ with pytest.raises(ValueError, match="report path"):
+ replace(_candidate(), security_report_path=None)
diff --git a/tests/fixtures/fake_terraform/terraform b/tests/fixtures/fake_terraform/terraform
index 329d163..83bc364 100755
--- a/tests/fixtures/fake_terraform/terraform
+++ b/tests/fixtures/fake_terraform/terraform
@@ -6,6 +6,7 @@ script covers every scenario instead of one file per case. Never touches
real infrastructure; only prints canned output and writes a placeholder
plan file when asked to `-out=`.
"""
+
from __future__ import annotations
import json
@@ -19,7 +20,7 @@ SCENARIO = os.environ.get("FAKE_TERRAFORM_SCENARIO", "success")
def _plan_out_path(args: list[str]) -> str | None:
for arg in args:
if arg.startswith("-out="):
- return arg[len("-out="):]
+ return arg[len("-out=") :]
return None
@@ -35,7 +36,7 @@ def main() -> int:
if subcommand == "fmt":
if SCENARIO == "fmt_needs_formatting":
- print("main.tf\n+ location = \"eastus\"", file=sys.stdout)
+ print('main.tf\n+ location = "eastus"', file=sys.stdout)
return 3
return 0
@@ -48,10 +49,16 @@ def main() -> int:
if subcommand == "validate":
if SCENARIO == "validate_failure":
- print(json.dumps({
- "valid": False,
- "diagnostics": [{"severity": "error", "summary": "Missing required argument"}],
- }))
+ print(
+ json.dumps(
+ {
+ "valid": False,
+ "diagnostics": [
+ {"severity": "error", "summary": "Missing required argument"}
+ ],
+ }
+ )
+ )
return 1
print(json.dumps({"valid": True, "diagnostics": []}))
return 0
@@ -80,47 +87,81 @@ def main() -> int:
print("{not valid json")
return 0
if SCENARIO == "plan_with_sensitive_attribute":
- print(json.dumps({
- "resource_changes": [
- {
- "address": "azurerm_resource_group.main",
- "change": {"actions": ["create"]},
- },
+ print(
+ json.dumps(
{
- "address": "azurerm_key_vault_secret.admin",
- "change": {
- "actions": ["create"],
- "after": {"admin_password": "hunter2"},
- },
- },
- ]
- }))
+ "resource_changes": [
+ {
+ "address": "azurerm_resource_group.main",
+ "change": {"actions": ["create"]},
+ },
+ {
+ "address": "azurerm_key_vault_secret.admin",
+ "change": {
+ "actions": ["create"],
+ "after": {"admin_password": "hunter2"},
+ },
+ },
+ ]
+ }
+ )
+ )
return 0
if SCENARIO == "plan_mixed_actions":
- print(json.dumps({
- "resource_changes": [
- {"address": "azurerm_resource_group.main",
- "change": {"actions": ["create"]}},
- {"address": "azurerm_container_registry.main",
- "change": {"actions": ["update"]}},
- {"address": "azurerm_log_analytics_workspace.main",
- "change": {"actions": ["delete", "create"]}},
- {"address": "azurerm_storage_account.old",
- "change": {"actions": ["delete"]}},
- {"address": "azurerm_resource_group.untouched",
- "change": {"actions": ["no-op"]}},
- ]
- }))
+ print(
+ json.dumps(
+ {
+ "resource_changes": [
+ {
+ "address": "azurerm_resource_group.main",
+ "change": {"actions": ["create"]},
+ },
+ {
+ "address": "azurerm_container_registry.main",
+ "change": {"actions": ["update"]},
+ },
+ {
+ "address": "azurerm_log_analytics_workspace.main",
+ "change": {"actions": ["delete", "create"]},
+ },
+ {
+ "address": "azurerm_storage_account.old",
+ "change": {"actions": ["delete"]},
+ },
+ {
+ "address": "azurerm_resource_group.untouched",
+ "change": {"actions": ["no-op"]},
+ },
+ ]
+ }
+ )
+ )
return 0
# default "success" scenario: two creates
- print(json.dumps({
- "resource_changes": [
- {"address": "azurerm_resource_group.main",
- "change": {"actions": ["create"]}},
- {"address": "azurerm_container_registry.main",
- "change": {"actions": ["create"]}},
- ]
- }))
+ print(
+ json.dumps(
+ {
+ "resource_changes": [
+ {
+ "address": "azurerm_resource_group.main",
+ "change": {"actions": ["create"]},
+ },
+ {
+ "address": "azurerm_container_registry.main",
+ "change": {"actions": ["create"]},
+ },
+ ]
+ }
+ )
+ )
+ return 0
+
+ if subcommand in {"apply", "destroy"}:
+ print(f"{subcommand.title()} complete!")
+ return 0
+
+ if subcommand == "output":
+ print(json.dumps({}))
return 0
return 1
diff --git a/tests/tools/test_azure_tool.py b/tests/tools/test_azure_tool.py
new file mode 100644
index 0000000..8786aca
--- /dev/null
+++ b/tests/tools/test_azure_tool.py
@@ -0,0 +1,70 @@
+from typing import Any, Mapping
+
+from devops_learn.tools.approval import (
+ ApprovalRecord,
+ AutoApproveApprovalGate,
+ AutoDenyApprovalGate,
+)
+from devops_learn.tools.azure_tool import _container_app_view
+from devops_learn.tools.base import RiskLevel, Tool, ToolOperationSpec, ToolResult
+from devops_learn.tools.service import ToolService
+
+
+def test_container_app_observation_is_sanitized_to_deployment_facts() -> None:
+ observed = _container_app_view(
+ {
+ "name": "api",
+ "location": "eastus",
+ "tags": {"environment": "learning"},
+ "properties": {
+ "latestRevisionName": "api--one",
+ "runningStatus": "Running",
+ "configuration": {"ingress": {"fqdn": "api.example.invalid"}},
+ "template": {"containers": [{"image": "acr.azurecr.io/api@sha256:abc"}]},
+ },
+ }
+ )
+ assert observed == {
+ "name": "api",
+ "location": "eastus",
+ "tags": {"environment": "learning"},
+ "image": "acr.azurecr.io/api@sha256:abc",
+ "endpoint": "api.example.invalid",
+ "revision": "api--one",
+ "running_status": "Running",
+ }
+
+
+def test_destroy_operation_requires_toolservice_approval() -> None:
+ denied = ToolService({"terraform": _DestroySpy()}, AutoDenyApprovalGate()).invoke(
+ "terraform", "destroy_approved_environment", {"environment": "learning"}
+ )
+ assert not denied.success
+ approved = ToolService({"terraform": _DestroySpy()}, AutoApproveApprovalGate()).invoke(
+ "terraform", "destroy_approved_environment", {"environment": "learning"}
+ )
+ assert approved.success
+
+
+class _DestroySpy(Tool):
+ @property
+ def name(self) -> str:
+ return "terraform"
+
+ @property
+ def operations(self) -> tuple[ToolOperationSpec, ...]:
+ return (
+ ToolOperationSpec(
+ "destroy_approved_environment", RiskLevel.DESTRUCTIVE, False, True, True
+ ),
+ )
+
+ def execute(
+ self,
+ operation: str,
+ params: Mapping[str, Any],
+ *,
+ dry_run: bool,
+ approval: ApprovalRecord | None,
+ ) -> ToolResult:
+ return ToolResult(True, "destroyed", {}, RiskLevel.DESTRUCTIVE, dry_run, approval)
diff --git a/tests/tools/test_real_terraform_tool.py b/tests/tools/test_real_terraform_tool.py
index 24ec66a..a429047 100644
--- a/tests/tools/test_real_terraform_tool.py
+++ b/tests/tools/test_real_terraform_tool.py
@@ -2,8 +2,7 @@
import os
from pathlib import Path
-import pytest
-
+from devops_learn.tools.approval import ApprovalRecord
from devops_learn.tools.base import RiskLevel
from devops_learn.tools.terraform_tool import RealTerraformTool
from devops_learn.validation import terraform_plan_analysis
@@ -16,22 +15,31 @@ def _use_fake_terraform(monkeypatch, scenario: str = "success") -> None:
monkeypatch.setenv("FAKE_TERRAFORM_SCENARIO", scenario)
-def test_real_terraform_tool_has_safe_operations() -> None:
+def test_real_terraform_tool_declares_real_apply_and_destroy_risk() -> None:
tool = RealTerraformTool()
assert tool.name == "terraform"
names = {spec.name for spec in tool.operations}
- assert names == {"fmt", "init", "validate", "plan"}
+ assert names == {
+ "fmt",
+ "init",
+ "validate",
+ "plan",
+ "output",
+ "apply_approved_plan",
+ "destroy_approved_environment",
+ }
for spec in tool.operations:
+ if spec.name in {"apply_approved_plan", "destroy_approved_environment"}:
+ assert spec.requires_approval
+ continue
assert spec.risk_level is RiskLevel.SAFE
assert not spec.requires_approval
-def test_real_terraform_tool_does_not_expose_apply_or_destroy() -> None:
+def test_real_terraform_tool_apply_and_destroy_require_approval() -> None:
tool = RealTerraformTool()
- with pytest.raises(KeyError):
- tool.spec_for("apply_approved_plan")
- with pytest.raises(KeyError):
- tool.spec_for("destroy_approved_environment")
+ assert tool.spec_for("apply_approved_plan").risk_level is RiskLevel.HIGH
+ assert tool.spec_for("destroy_approved_environment").risk_level is RiskLevel.DESTRUCTIVE
def test_missing_binary_reports_failure(monkeypatch, tmp_path) -> None:
@@ -180,6 +188,69 @@ def test_invalid_working_directory_reports_failure(monkeypatch, tmp_path) -> Non
assert not result.success
+def test_saved_plan_binds_digest_source_config_and_candidate(monkeypatch, tmp_path) -> None:
+ _use_fake_terraform(monkeypatch)
+ tool = RealTerraformTool()
+ planned = tool.execute(
+ "plan",
+ {"path": str(tmp_path), "source_revision": "source-a", "candidate_context": "candidate-a"},
+ dry_run=False,
+ approval=None,
+ )
+ assert planned.success
+ assert Path(str(planned.details["plan_path"])).is_file()
+ assert planned.details["plan_digest"]
+ approved = ApprovalRecord(granted=True, approved_by="test")
+ applied = tool.execute(
+ "apply_approved_plan",
+ {
+ "path": str(tmp_path),
+ "plan_path": planned.details["plan_path"],
+ "source_revision": "source-a",
+ "candidate_context": "candidate-a",
+ },
+ dry_run=False,
+ approval=approved,
+ )
+ assert applied.success
+
+
+def test_apply_refuses_changed_source_or_plan(monkeypatch, tmp_path) -> None:
+ _use_fake_terraform(monkeypatch)
+ tool = RealTerraformTool()
+ planned = tool.execute(
+ "plan",
+ {"path": str(tmp_path), "source_revision": "source-a", "candidate_context": "candidate-a"},
+ dry_run=False,
+ approval=None,
+ )
+ changed_source = tool.execute(
+ "apply_approved_plan",
+ {
+ "path": str(tmp_path),
+ "plan_path": planned.details["plan_path"],
+ "source_revision": "source-b",
+ "candidate_context": "candidate-a",
+ },
+ dry_run=False,
+ approval=ApprovalRecord(granted=True, approved_by="test"),
+ )
+ assert not changed_source.success
+ Path(str(planned.details["plan_path"])).write_bytes(b"changed")
+ changed_plan = tool.execute(
+ "apply_approved_plan",
+ {
+ "path": str(tmp_path),
+ "plan_path": planned.details["plan_path"],
+ "source_revision": "source-a",
+ "candidate_context": "candidate-a",
+ },
+ dry_run=False,
+ approval=ApprovalRecord(granted=True, approved_by="test"),
+ )
+ assert not changed_plan.success
+
+
def test_dry_run_never_invokes_subprocess(monkeypatch, tmp_path) -> None:
monkeypatch.setenv("PATH", str(tmp_path)) # empty dir: would fail loudly if invoked
result = RealTerraformTool().execute(