Título Zero Trust Network Access (ZTNA): Substituindo VPNs Corporativas por AWS Verified Access e IAM Identity Center

resource "aws_verifiedaccess_trust_provider" "identity_center" {
policy_reference_name = "idc"
trust_provider_type = "user"
user_trust_provider_type = "iam-identity-center"

tags = {
Environment = "Production"
Domain = "Security"
}
}

resource "aws_verifiedaccess_trust_provider" "device_posture" {
policy_reference_name = "crowdstrike"
trust_provider_type = "device"
device_trust_provider_type = "crowdstrike"

device_options {
tenant_id = var.crowdstrike_tenant_id
}
}

resource "aws_verifiedaccess_group" "finance_applications" {
verifiedaccess_instance_id = aws_verifiedaccess_instance.main.id

policy_document = jsonencode({
"CedarJson": "permit(principal, action, resource) when { context.idc.user.email.contains(\"@enterprise.com\") && context.crowdstrike.risk_score < 30 };"
})

tags = {
RiskTier = "High"
}
}

resource "aws_verifiedaccess_endpoint" "finance_dashboard" {
application_domain = "finance.enterprise.com"
attachment_type = "vpc"
domain_certificate_arn = aws_acm_certificate.public_cert.arn
endpoint_domain_prefix = "finance-dash"
endpoint_type = "load-balancer"
verified_access_group_id = aws_verifiedaccess_group.finance_applications.id

load_balancer_options {
protocol = "https"
port = 443
load_balancer_arn = aws_lb.internal_finance_alb.arn
subnet_ids = [var.private_subnet_1, var.private_subnet_2]
}

security_group_ids = [aws_security_group.ava_egress.id]
}

import os
import json
import jwt
import requests
from typing import Dict, Any
from aws_lambda_powertools import Logger

logger = Logger(service="FinanceAdapter")

class VerifiedAccessAdapter:
def __init__(self, region: str, endpoint_id: str):
# A chave pública é imutável e fornecida pela AWS para verificar o JWT gerado pelo AVA
self.public_key_url = f"https://public-keys.auth.elb.{region}.amazonaws.com/{endpoint_id}"
self.public_key = None

def _fetch_public_key(self) -> str:
if not self.public_key:
response = requests.get(self.public_key_url, timeout=2.0)
response.raise_for_status()
self.public_key = response.text
return self.public_key

def decode_and_verify_context(self, headers: Dict[str, str]) -> Dict[str, Any]:
jwt_token = headers.get('x-amzn-ava-user-context')
if not jwt_token:
raise PermissionError("Cabeçalho de contexto Zero Trust ausente.")

public_key = self._fetch_public_key()

try:
# A decodificação matemática garante a proveniência intransponível do dado
decoded_claims = jwt.decode(
jwt_token,
public_key,
algorithms=["ES256"],
options={"verify_signature": True, "verify_exp": True}
)
logger.info(f"Contexto decodificado com sucesso para {decoded_claims.get('email')}")
return decoded_claims

except jwt.PyJWTError as e:
logger.error(f"Violação de assinatura JWT interceptada: {str(e)}")
raise PermissionError("Invalidez matemática no atestado de identidade.")

ava_adapter = VerifiedAccessAdapter(region="us-east-1", endpoint_id="ve-0abcd1234efgh5678")

@logger.inject_lambda_context
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
# Extração de cabeçalhos normalizados (em minúsculas)
headers = {k.lower(): v for k, v in event.get('headers', {}).items()}

try:
user_context = ava_adapter.decode_and_verify_context(headers)
return {
"statusCode": 200,
"body": json.dumps({
"status": "Acesso Autorizado",
"corporate_identity": user_context.get("email")
})
}
except PermissionError as e:
return {"statusCode": 401, "body": json.dumps({"error": str(e)})}