Usage examples

from qwed_infra import TerraformParser

parser = TerraformParser()
resources = parser.parse_directory("./terraform/prod")

# Returns normalized resources:
# {
#   "policies": [...],
#   "instances": [...],
#   "subnets": [...],
#   "route_tables": [...],
#   "security_groups": {...},
#   ...
# }

from qwed_infra import IamGuard

guard = IamGuard()

# policy comes from parsed Terraform resources (see Example 1)
policy = resources["policies"][0]

# Context-aware access check
result = guard.verify_access(
    policy,
    action="s3:GetObject",
    resource="arn:aws:s3:::bucket/*",
    context={"aws:SourceIp": "192.168.1.5"},
)
print(f"Allowed? {result.allowed}")   # True/False
print(f"Verified? {result.verified}") # True/False
print(f"Proof: {result.proof}")       # Z3 proof string

# Least-privilege check
result = guard.verify_least_privilege(policy)
print(f"Over-privileged? {result.allowed}")

from qwed_infra import CostGuard

cost = CostGuard()

resources = {
    "instances": [
        {"id": "web-cluster", "instance_type": "t3.micro", "count": 2},
        {"id": "gpu-trainer", "instance_type": "p4d.24xlarge", "count": 1},
    ]
}

result = cost.verify_budget(resources, budget_monthly=500.0)

print(f"Within Budget? {result.within_budget}")  # False
print(f"Total: ${result.total_monthly_cost}")     # ~23905.36
print(f"Reason: {result.reason}")

from qwed_infra import NetworkGuard

net = NetworkGuard()

infra = {
    "subnets": [
        {"id": "public-subnet", "security_groups": ["sg-web"]}
    ],
    "route_tables": [
        {"subnet_id": "public-subnet", "routes": {"0.0.0.0/0": "igw-main"}}
    ],
    "security_groups": {
        "sg-web": {"ingress": [{"port": 80, "cidr": "0.0.0.0/0"}]}
    },
}

# Is port 80 accessible from the internet?
res = net.verify_reachability(infra, "internet", "public-subnet", 80)
print(f"Reachable? {res.reachable}")  # True
print(f"Path: {res.path}")           # ['internet', 'igw-main', ...]

from pathlib import Path
from qwed_infra import ArtifactBoundaryGuard

guard = ArtifactBoundaryGuard()

result = guard.verify_package_boundary(
    package_dir=Path("./src/qwed_infra"),
    pyproject_path=Path("./pyproject.toml"),
    package_name="qwed_infra",
)

print(f"Safe to publish? {result.is_safe}")
print(f"Findings: {len(result.findings)}")
for f in result.findings:
    print(f"  [{f.severity}] {f.finding_type}: {f.file_path} — {f.reason}")

# From IamGuard
diag = IamGuard.to_diagnostic(result)  # returns InfraDiagnosticResult

print(f"Status: {diag.status.value}")         # VERIFIED / UNVERIFIABLE / BLOCKED
print(f"Agent: {diag.agent_message}")          # Layer 1
print(f"Constraint: {diag.constraint_id}")     # Layer 2 field
print(f"Authoritative: {diag.is_authoritative}")  # True iff proof_ref is set
print(f"Proof: {diag.proof_ref}")              # Layer 3 — sha256:... or None

# Serialize for audit logging
payload = diag.to_dict()

# Deserialize back
from qwed_infra.diagnostics import InfraDiagnosticResult
restored = InfraDiagnosticResult.from_dict(payload)

from qwed_infra.audit import (
    IAM_DENY_PRECEDENCE,
    NETWORK_REACHABILITY,
    COST_WITHIN_BUDGET,
    ARTIFACT_BOUNDARY_VERIFIED,
    build_trace,
    trace_proof_ref,
)

# Build a structured audit trace
trace = build_trace(
    IAM_DENY_PRECEDENCE,
    outcome="ALLOWED",
    inputs={"action": "s3:GetObject", "resource": "arn:aws:s3:::bucket/*"},
)
# trace = {
#     "rule_id": "IAM_DENY_PRECEDENCE",
#     "statute": "AWS IAM Evaluation Logic (deny precedence)",
#     "jurisdiction": "GENERIC",
#     "outcome": "ALLOWED",
#     "inputs": {"action": "s3:GetObject", "resource": "..."},
# }

# Compute proof reference from a trace
ref = trace_proof_ref(trace)
print(ref)  # sha256:abc123...

from qwed_infra import NetworkGuard
from qwed_infra.attestation import mint_diagnostic_attestation

net = NetworkGuard()
infra = {
    "subnets": [{"id": "public-subnet", "security_groups": ["sg-web"]}],
    "route_tables": [
        {"subnet_id": "public-subnet", "routes": {"0.0.0.0/0": "igw-main"}}
    ],
    "security_groups": {
        "sg-web": {"ingress": [{"port": 80, "cidr": "0.0.0.0/0"}]}
    },
}

statement = "Traffic from internet to public-subnet on port 80 is safe"

# Mint an attestation bound to this claim + evidence
diagnostic = NetworkGuard.to_diagnostic(
    net.verify_reachability(infra, "internet", "public-subnet", 80)
)
attestation = mint_diagnostic_attestation(
    diagnostic, engine="NetworkGuard", query=statement
)
if not attestation.is_issued:
    raise RuntimeError(f"Attestation unavailable [{attestation.error_code}]")

# The guard re-runs the reachability check internally — you pass raw inputs,
# never a pre-computed result object
doc = net.to_verification_context(
    infra,
    "internet",
    "public-subnet",
    80,
    formal_statement=statement,
    attestation_token=attestation.token,
)

print(doc.verdict.value)                     # VERIFIED
print(doc.context.decision.admission.value)  # ADMIT

from qwed_infra.attestation import get_attestation_service
from qwed_infra.verification_context import is_valid_document, resolve_document_proof_ref

document = doc.to_dict()  # JSON-serializable dict
token = attestation.token  # the attestation minted alongside the document

if not is_valid_document(document):
    raise ValueError("Invalid VC document — reject")

admission = document["context"]["decision"]["admission"]  # ADMIT or DENY

if admission == "ADMIT":
    # VERIFIED documents carry a proof_ref bound to the exact evidence
    if not resolve_document_proof_ref(document):
        raise ValueError("VC document proof_ref does not resolve — reject")

    # ADMIT is only trustworthy with a valid attestation bound to the same
    # claim and evidence. Verify the token's signature, issuer, expiry, and
    # revocation status BEFORE honoring the admission — a structurally valid
    # document with a forged or expired attestation must never reach the gate.
    service = get_attestation_service()
    ok, claims, error = service.verify_attestation(token)
    if not ok:
        raise PermissionError(f"Attestation verification failed: {error} — reject")

    # ... proceed with the gated operation ...
else:
    raise PermissionError(f"Verification denied ({document['verdict']}) — reject")