File size: 13,585 Bytes
517919c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 | """CloudFront in RedactionStack: magic-link auth and distribution synth."""
import sys
from pathlib import Path
CDK_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(CDK_DIR))
from cdk_cloudfront_auth import (
build_forwarded_host_viewer_request_js,
build_magic_link_viewer_request_js,
)
from cdk_cloudfront_distribution import (
express_endpoint_hostname,
parse_geo_restriction_locations,
)
def test_express_endpoint_hostname():
assert (
express_endpoint_hostname("https://abc123.eu-west-2.on.aws")
== "abc123.eu-west-2.on.aws"
)
assert express_endpoint_hostname("abc123.eu-west-2.on.aws") == (
"abc123.eu-west-2.on.aws"
)
def test_parse_geo_restriction_locations():
assert parse_geo_restriction_locations("GB") == ["GB"]
assert parse_geo_restriction_locations("gb, us") == ["GB", "US"]
assert parse_geo_restriction_locations("") is None
def test_magic_link_viewer_request_js_embeds_token_and_cookie():
js = build_magic_link_viewer_request_js(
token="a1b2c3d4e5f6789012345678901234ab",
cookie_name="doc-redaction-auth",
cookie_max_age_sec=604800,
)
assert "a1b2c3d4e5f6789012345678901234ab" in js
assert "doc-redaction-auth" in js
assert "604800" in js
assert "statusCode: 302" in js
assert "statusCode: 401" in js
def test_magic_link_viewer_request_js_forwards_viewer_host():
"""Authorized requests must forward the viewer host so Gradio builds asset URLs
against the CloudFront domain, not the *.on.aws origin host. The proto is NOT set
here: x-forwarded-proto is a CloudFront-disallowed edge-function header (HTTP 502).
"""
js = build_magic_link_viewer_request_js(
token="a1b2c3d4e5f6789012345678901234ab",
cookie_name="doc-redaction-auth",
cookie_max_age_sec=604800,
)
assert "x-forwarded-host" in js
assert "x-forwarded-proto" not in js
def test_forwarded_host_viewer_request_js_sets_headers():
js = build_forwarded_host_viewer_request_js()
assert "x-forwarded-host" in js
# x-forwarded-proto is CloudFront-disallowed for edge functions; supplied via a
# static custom origin header instead (see distribution config).
assert "x-forwarded-proto" not in js
assert "request.headers.host.value" in js
assert "return request;" in js
def test_cloudfront_without_magic_link_still_forwards_host():
"""auth_mode='none' must still attach a viewer-request function that forwards
the viewer host (otherwise Gradio assets 404/time out behind CloudFront)."""
template = _synth_cloudfront_with_headers(attach=False)
template.resource_count_is("AWS::CloudFront::Function", 1)
def test_express_cloudfront_synth_no_waf():
from aws_cdk import App, Environment, Stack, assertions
from cdk_cloudfront_distribution import create_redaction_cloudfront_distribution
from cdk_functions import managed_resource_removal_policy
app = App()
stack = Stack(
app,
"ExpressCloudFrontTest",
env=Environment(account="123456789012", region="eu-west-2"),
)
create_redaction_cloudfront_distribution(
stack,
"Cf",
distribution_comment="test-dist",
cognito_redirection_url="https://main.example.on.aws",
cloudfront_domain="d111.cloudfront.net",
cognito_user_pool_domain_prefix="demo",
aws_region="eu-west-2",
cognito_user_pool_login_url="",
ssl_certificate_domain="",
enable_secure_response_headers=False,
geo_restriction_raw="GB",
enable_cloudfront_waf=False,
web_acl_name="test-waf",
auth_mode="magic-link",
magic_link_cookie_name="doc-redaction-auth",
magic_link_cookie_max_age_sec=604800,
custom_header_name="",
custom_header_value="",
cdk_prefix="Test",
resource_removal_policy=managed_resource_removal_policy(),
main_express_endpoint="https://main.example.on.aws",
agentic_express_endpoint="https://agentic.example.on.aws",
agentic_path_prefix="/agent",
)
template = assertions.Template.from_stack(stack)
template.resource_count_is("AWS::WAFv2::WebACL", 0)
template.resource_count_is("AWS::CloudFront::Distribution", 1)
template.resource_count_is("AWS::CloudFront::Function", 1)
resources = template.to_json()["Resources"]
dist = next(
r for r in resources.values() if r["Type"] == "AWS::CloudFront::Distribution"
)
dist_config = dist["Properties"]["DistributionConfig"]
behaviors = dist_config["CacheBehaviors"]
assert len(behaviors) >= 2
# Express origins must NOT forward the viewer Host header: the ECS managed ALB routes
# on each service's own *.on.aws host, so CloudFront sends the origin domain as Host.
# ALL_VIEWER_EXCEPT_HOST_HEADER managed policy id (stable AWS constant).
all_viewer_except_host = "b689b0a8-53d0-40ab-baf2-68738e2966ac"
assert (
dist_config["DefaultCacheBehavior"]["OriginRequestPolicyId"]
== all_viewer_except_host
)
for behavior in behaviors:
assert behavior["OriginRequestPolicyId"] == all_viewer_except_host
def test_magic_link_emits_agent_app_login_url_output():
"""When the agent (Pi) app has a dedicated CloudFront behavior, the stack must
expose AgentRedactionUrl / AgentRedactionLoginUrl so operators can unlock and
open the agent UI directly (magic-link cookie is domain-wide)."""
from aws_cdk import App, Environment, Stack, assertions
from cdk_cloudfront_distribution import create_redaction_cloudfront_distribution
from cdk_functions import managed_resource_removal_policy
app = App()
stack = Stack(
app,
"AgentLoginUrlTest",
env=Environment(account="123456789012", region="eu-west-2"),
)
create_redaction_cloudfront_distribution(
stack,
"Cf",
distribution_comment="test-dist",
cognito_redirection_url="https://main.example.on.aws",
cloudfront_domain="d111.cloudfront.net",
cognito_user_pool_domain_prefix="demo",
aws_region="eu-west-2",
cognito_user_pool_login_url="",
ssl_certificate_domain="",
enable_secure_response_headers=False,
geo_restriction_raw="GB",
enable_cloudfront_waf=False,
web_acl_name="test-waf",
auth_mode="magic-link",
magic_link_cookie_name="doc-redaction-auth",
magic_link_cookie_max_age_sec=604800,
custom_header_name="",
custom_header_value="",
cdk_prefix="Test",
resource_removal_policy=managed_resource_removal_policy(),
main_express_endpoint="https://main.example.on.aws",
agentic_express_endpoint="https://agentic.example.on.aws",
agentic_path_prefix="/agent",
)
outputs = assertions.Template.from_stack(stack).to_json().get("Outputs", {})
assert "AgentRedactionUrl" in outputs
assert "AgentRedactionLoginUrl" in outputs
def test_no_agent_output_when_agent_shares_main_origin():
"""No separate agent behavior (agentic host == main host) -> no agent URL output."""
from aws_cdk import App, Environment, Stack, assertions
from cdk_cloudfront_distribution import create_redaction_cloudfront_distribution
from cdk_functions import managed_resource_removal_policy
app = App()
stack = Stack(
app,
"NoAgentOutputTest",
env=Environment(account="123456789012", region="eu-west-2"),
)
create_redaction_cloudfront_distribution(
stack,
"Cf",
distribution_comment="test-dist",
cognito_redirection_url="https://main.example.on.aws",
cloudfront_domain="d111.cloudfront.net",
cognito_user_pool_domain_prefix="demo",
aws_region="eu-west-2",
cognito_user_pool_login_url="",
ssl_certificate_domain="",
enable_secure_response_headers=False,
geo_restriction_raw="GB",
enable_cloudfront_waf=False,
web_acl_name="test-waf",
auth_mode="magic-link",
magic_link_cookie_name="doc-redaction-auth",
magic_link_cookie_max_age_sec=604800,
custom_header_name="",
custom_header_value="",
cdk_prefix="Test",
resource_removal_policy=managed_resource_removal_policy(),
main_express_endpoint="https://main.example.on.aws",
agentic_express_endpoint="",
agentic_path_prefix="/agent",
)
outputs = assertions.Template.from_stack(stack).to_json().get("Outputs", {})
assert "RedactionLoginUrl" in outputs
assert "AgentRedactionUrl" not in outputs
assert "AgentRedactionLoginUrl" not in outputs
def test_express_origins_set_static_forwarded_proto_header():
"""Every Express origin must carry a static X-Forwarded-Proto: https custom header
so Gradio emits https asset URLs (the edge function can't set this header)."""
from aws_cdk import App, Environment, Stack, assertions
from cdk_cloudfront_distribution import create_redaction_cloudfront_distribution
from cdk_functions import managed_resource_removal_policy
app = App()
stack = Stack(
app,
"ExpressFwdProtoTest",
env=Environment(account="123456789012", region="eu-west-2"),
)
create_redaction_cloudfront_distribution(
stack,
"Cf",
distribution_comment="test-dist",
cognito_redirection_url="https://main.example.on.aws",
cloudfront_domain="d111.cloudfront.net",
cognito_user_pool_domain_prefix="demo",
aws_region="eu-west-2",
cognito_user_pool_login_url="",
ssl_certificate_domain="",
enable_secure_response_headers=False,
geo_restriction_raw="GB",
enable_cloudfront_waf=False,
web_acl_name="test-waf",
auth_mode="magic-link",
magic_link_cookie_name="doc-redaction-auth",
magic_link_cookie_max_age_sec=604800,
custom_header_name="",
custom_header_value="",
cdk_prefix="Test",
resource_removal_policy=managed_resource_removal_policy(),
main_express_endpoint="https://main.example.on.aws",
agentic_express_endpoint="https://agentic.example.on.aws",
agentic_path_prefix="/agent",
)
template = assertions.Template.from_stack(stack)
dist = next(
r
for r in template.to_json()["Resources"].values()
if r["Type"] == "AWS::CloudFront::Distribution"
)
origins_cfg = dist["Properties"]["DistributionConfig"]["Origins"]
assert origins_cfg, "expected at least one origin"
for origin in origins_cfg:
header_pairs = {
h["HeaderName"]: h["HeaderValue"]
for h in origin.get("OriginCustomHeaders", [])
}
assert header_pairs.get("X-Forwarded-Proto") == "https"
def _synth_cloudfront_with_headers(attach: bool):
from aws_cdk import App, Environment, Stack, assertions
from cdk_cloudfront_distribution import create_redaction_cloudfront_distribution
from cdk_functions import managed_resource_removal_policy
app = App()
stack = Stack(
app,
f"HeadersTest{'Attach' if attach else 'Detach'}",
env=Environment(account="123456789012", region="eu-west-2"),
)
create_redaction_cloudfront_distribution(
stack,
"Cf",
distribution_comment="test-dist",
cognito_redirection_url="https://main.example.on.aws",
cloudfront_domain="d111.cloudfront.net",
cognito_user_pool_domain_prefix="demo",
aws_region="eu-west-2",
cognito_user_pool_login_url="",
ssl_certificate_domain="",
enable_secure_response_headers=True,
attach_secure_response_headers=attach,
geo_restriction_raw="GB",
enable_cloudfront_waf=False,
web_acl_name="test-waf",
auth_mode="none",
magic_link_cookie_name="doc-redaction-auth",
magic_link_cookie_max_age_sec=604800,
custom_header_name="",
custom_header_value="",
cdk_prefix="Test",
resource_removal_policy=managed_resource_removal_policy(),
main_express_endpoint="https://main.example.on.aws",
agentic_express_endpoint="https://agentic.example.on.aws",
agentic_path_prefix="/agent",
)
return assertions.Template.from_stack(stack)
def _behavior_response_headers_ids(template) -> list:
dist = next(
r
for r in template.to_json()["Resources"].values()
if r["Type"] == "AWS::CloudFront::Distribution"
)
config = dist["Properties"]["DistributionConfig"]
behaviors = [config["DefaultCacheBehavior"], *config.get("CacheBehaviors", [])]
return [b.get("ResponseHeadersPolicyId") for b in behaviors]
def test_response_headers_policy_created_but_detached_by_default():
template = _synth_cloudfront_with_headers(attach=False)
# The policy resource is created (appears in the account's policy list)...
template.resource_count_is("AWS::CloudFront::ResponseHeadersPolicy", 1)
# ...but is not attached to any distribution behavior.
ids = _behavior_response_headers_ids(template)
assert all(policy_id is None for policy_id in ids), ids
def test_response_headers_policy_attached_when_requested():
template = _synth_cloudfront_with_headers(attach=True)
template.resource_count_is("AWS::CloudFront::ResponseHeadersPolicy", 1)
ids = _behavior_response_headers_ids(template)
assert ids, ids
assert all(policy_id is not None for policy_id in ids), ids
|