【検証】VPCからBrave Searchを使う：Network Firewallで api.search.brave.com だけ穴を開けて、Web検索・AI回答が通るか試す

#!/usr/bin/env node
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import {
  aws_ec2 as ec2,
  aws_iam as iam,
  aws_logs as logs,
  aws_networkfirewall as netfw,
  aws_secretsmanager as secretsmanager,
} from 'aws-cdk-lib';

// 教材用パラメータ（必要に応じて変更する）
const VPC_CIDR = '10.0.0.0/16';
const WORKLOAD_CIDR = '10.0.1.0/24'; // 検証用 EC2（プライベート）
const PUBLIC_CIDR = '10.0.0.0/24'; // NAT Gateway（パブリック）
const FIREWALL_CIDR = '10.0.2.0/24'; // Network Firewall エンドポイント
const SECRET_NAME = 'brave-search/api-key';
const FIREWALL_NAME = 'brave-egress-fw';

class BraveNfwEgressStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const az = cdk.Fn.select(0, cdk.Fn.getAzs());

    // 1. VPC と Internet Gateway
    const vpc = new ec2.CfnVPC(this, 'Vpc', {
      cidrBlock: VPC_CIDR,
      enableDnsSupport: true,
      enableDnsHostnames: true,
      tags: [{ key: 'Name', value: 'brave-nfw-egress-vpc' }],
    });
    const igw = new ec2.CfnInternetGateway(this, 'Igw', {
      tags: [{ key: 'Name', value: 'brave-nfw-igw' }],
    });
    const igwAttach = new ec2.CfnVPCGatewayAttachment(this, 'IgwAttach', {
      vpcId: vpc.ref,
      internetGatewayId: igw.ref,
    });

    // 2. サブネット3層（すべて単一AZ）
    const workloadSubnet = new ec2.CfnSubnet(this, 'WorkloadSubnet', {
      vpcId: vpc.ref, cidrBlock: WORKLOAD_CIDR, availabilityZone: az,
      mapPublicIpOnLaunch: false, tags: [{ key: 'Name', value: 'workload-private' }],
    });
    const publicSubnet = new ec2.CfnSubnet(this, 'PublicSubnet', {
      vpcId: vpc.ref, cidrBlock: PUBLIC_CIDR, availabilityZone: az,
      mapPublicIpOnLaunch: false, tags: [{ key: 'Name', value: 'nat-public' }],
    });
    const firewallSubnet = new ec2.CfnSubnet(this, 'FirewallSubnet', {
      vpcId: vpc.ref, cidrBlock: FIREWALL_CIDR, availabilityZone: az,
      mapPublicIpOnLaunch: false, tags: [{ key: 'Name', value: 'firewall' }],
    });

    // 3. NAT Gateway（パブリックサブネットに配置）
    const natEip = new ec2.CfnEIP(this, 'NatEip', { domain: 'vpc' });
    const natgw = new ec2.CfnNatGateway(this, 'NatGw', {
      subnetId: publicSubnet.ref, allocationId: natEip.attrAllocationId,
      tags: [{ key: 'Name', value: 'brave-nfw-nat' }],
    });
    natgw.addDependency(igwAttach);

    // 4. Network Firewall（ポリシー＋本体）。初期はルールグループ無し＝全egress遮断。
    //    drop_strict ではなく drop_established（SNIを読むため接続確立を許す）。
    const policy = new netfw.CfnFirewallPolicy(this, 'FwPolicy', {
      firewallPolicyName: 'brave-egress-policy',
      firewallPolicy: {
        statelessDefaultActions: ['aws:forward_to_sfe'],
        statelessFragmentDefaultActions: ['aws:forward_to_sfe'],
        statefulEngineOptions: { ruleOrder: 'STRICT_ORDER' },
        statefulDefaultActions: ['aws:drop_established'],
      },
    });
    const firewall = new netfw.CfnFirewall(this, 'Firewall', {
      firewallName: FIREWALL_NAME,
      firewallPolicyArn: policy.attrFirewallPolicyArn,
      vpcId: vpc.ref,
      subnetMappings: [{ subnetId: firewallSubnet.ref }],
    });
    // attrEndpointIds は ["<az>:<vpce-id>"] 形式の配列で返る
    const fwEndpointId = cdk.Fn.select(1, cdk.Fn.split(':', cdk.Fn.select(0, firewall.attrEndpointIds)));

    // NFWフローログ
    const fwLogGroup = new logs.LogGroup(this, 'FwFlowLogs', {
      logGroupName: '/brave-nfw/flow',
      retention: logs.RetentionDays.ONE_WEEK,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });
    new netfw.CfnLoggingConfiguration(this, 'FwLogging', {
      firewallArn: firewall.attrFirewallArn,
      loggingConfiguration: {
        logDestinationConfigs: [
          { logType: 'FLOW', logDestinationType: 'CloudWatchLogs',
            logDestination: { logGroup: fwLogGroup.logGroupName } },
        ],
      },
    });

    // 5. ルートテーブルと経路
    // 5-1. workload → NAT
    const rtWorkload = new ec2.CfnRouteTable(this, 'RtWorkload', { vpcId: vpc.ref, tags: [{ key: 'Name', value: 'rt-workload' }] });
    new ec2.CfnSubnetRouteTableAssociation(this, 'AssocWorkload', { subnetId: workloadSubnet.ref, routeTableId: rtWorkload.ref });
    new ec2.CfnRoute(this, 'WorkloadDefaultRoute', { routeTableId: rtWorkload.ref, destinationCidrBlock: '0.0.0.0/0', natGatewayId: natgw.ref });
    // 5-2. public(NAT) → NFWエンドポイント
    const rtPublic = new ec2.CfnRouteTable(this, 'RtPublic', { vpcId: vpc.ref, tags: [{ key: 'Name', value: 'rt-nat-public' }] });
    new ec2.CfnSubnetRouteTableAssociation(this, 'AssocPublic', { subnetId: publicSubnet.ref, routeTableId: rtPublic.ref });
    new ec2.CfnRoute(this, 'PublicDefaultRoute', { routeTableId: rtPublic.ref, destinationCidrBlock: '0.0.0.0/0', vpcEndpointId: fwEndpointId });
    // 5-3. firewall → IGW
    const rtFirewall = new ec2.CfnRouteTable(this, 'RtFirewall', { vpcId: vpc.ref, tags: [{ key: 'Name', value: 'rt-firewall' }] });
    new ec2.CfnSubnetRouteTableAssociation(this, 'AssocFirewall', { subnetId: firewallSubnet.ref, routeTableId: rtFirewall.ref });
    const fwDefaultRoute = new ec2.CfnRoute(this, 'FirewallDefaultRoute', { routeTableId: rtFirewall.ref, destinationCidrBlock: '0.0.0.0/0', gatewayId: igw.ref });
    fwDefaultRoute.addDependency(igwAttach);
    // 5-4. IGW edge route（戻りをNFWへ）
    const rtIgwEdge = new ec2.CfnRouteTable(this, 'RtIgwEdge', { vpcId: vpc.ref, tags: [{ key: 'Name', value: 'rt-igw-edge' }] });
    const igwEdgeAssoc = new ec2.CfnGatewayRouteTableAssociation(this, 'AssocIgwEdge', { gatewayId: igw.ref, routeTableId: rtIgwEdge.ref });
    igwEdgeAssoc.addDependency(igwAttach);
    new ec2.CfnRoute(this, 'IgwEdgeReturnRoute', { routeTableId: rtIgwEdge.ref, destinationCidrBlock: PUBLIC_CIDR, vpcEndpointId: fwEndpointId });

    // 6. Secrets Manager（プレースホルダ）
    const braveSecret = new secretsmanager.Secret(this, 'BraveApiKey', {
      secretName: SECRET_NAME,
      description: 'Brave Search API key. Set via: aws secretsmanager put-secret-value',
    });
    braveSecret.applyRemovalPolicy(cdk.RemovalPolicy.DESTROY);

    // 7. VPCエンドポイント（ssm / ssmmessages / secretsmanager）
    const endpointSg = new ec2.CfnSecurityGroup(this, 'EndpointSg', {
      vpcId: vpc.ref,
      groupDescription: 'SG for interface endpoints (allow 443 from VPC)',
      securityGroupIngress: [{ ipProtocol: 'tcp', fromPort: 443, toPort: 443, cidrIp: VPC_CIDR, description: 'HTTPS from within VPC' }],
      tags: [{ key: 'Name', value: 'endpoint-sg' }],
    });
    for (const svc of ['ssm', 'ssmmessages', 'secretsmanager']) {
      new ec2.CfnVPCEndpoint(this, `Endpoint-${svc}`, {
        vpcId: vpc.ref,
        serviceName: `com.amazonaws.${this.region}.${svc}`,
        vpcEndpointType: 'Interface',
        subnetIds: [workloadSubnet.ref],
        securityGroupIds: [endpointSg.ref],
        privateDnsEnabled: true,
      });
    }

    // 8. 検証用 EC2（Amazon Linux 2023）。SGの description はASCIIのみ。
    const instanceSg = new ec2.CfnSecurityGroup(this, 'InstanceSg', {
      vpcId: vpc.ref,
      groupDescription: 'Workload SG: egress 443 + DNS only (domain control is on NFW)',
      securityGroupEgress: [
        { ipProtocol: 'tcp', fromPort: 443, toPort: 443, cidrIp: '0.0.0.0/0', description: 'HTTPS out (dest IP not restricted; domain control on NFW)' },
        { ipProtocol: 'udp', fromPort: 53, toPort: 53, cidrIp: VPC_CIDR, description: 'DNS' },
        { ipProtocol: 'tcp', fromPort: 53, toPort: 53, cidrIp: VPC_CIDR, description: 'DNS over TCP' },
      ],
      tags: [{ key: 'Name', value: 'workload-sg' }],
    });
    const role = new iam.Role(this, 'InstanceRole', {
      assumedBy: new iam.ServicePrincipal('ec2.amazonaws.com'),
      managedPolicies: [iam.ManagedPolicy.fromAwsManagedPolicyName('AmazonSSMManagedInstanceCore')],
    });
    braveSecret.grantRead(role);
    const instanceProfile = new iam.CfnInstanceProfile(this, 'InstanceProfile', { roles: [role.roleName] });
    const ami = ec2.MachineImage.latestAmazonLinux2023().getImage(this);
    const instance = new ec2.CfnInstance(this, 'WorkloadInstance', {
      imageId: ami.imageId,
      instanceType: 't3.micro',
      subnetId: workloadSubnet.ref,
      securityGroupIds: [instanceSg.ref],
      iamInstanceProfile: instanceProfile.ref,
      tags: [{ key: 'Name', value: 'brave-workload' }],
    });

    // 9. 出力
    new cdk.CfnOutput(this, 'FirewallName', { value: FIREWALL_NAME });
    new cdk.CfnOutput(this, 'FirewallPolicyArn', { value: policy.attrFirewallPolicyArn });
    new cdk.CfnOutput(this, 'InstanceId', { value: instance.ref });
    new cdk.CfnOutput(this, 'SecretName', { value: SECRET_NAME });
    new cdk.CfnOutput(this, 'SsmStartSession', { value: `aws ssm start-session --target ${instance.ref}` });
  }
}

const app = new cdk.App();
new BraveNfwEgressStack(app, 'BraveNfwEgressStack', {
  env: { account: process.env.CDK_DEFAULT_ACCOUNT, region: process.env.CDK_DEFAULT_REGION },
});
app.synth();

{
  "name": "brave-nfw-egress-cdk",
  "version": "1.0.0",
  "private": true,
  "scripts": { "synth": "cdk synth", "deploy": "cdk deploy", "destroy": "cdk destroy" },
  "devDependencies": {
    "@types/node": "^22.10.2",
    "aws-cdk": "^2.1000.0",
    "ts-node": "^10.9.2",
    "typescript": "~5.6.3"
  },
  "dependencies": { "aws-cdk-lib": "^2.170.0", "constructs": "^10.4.2" }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "commonjs",
    "lib": ["ES2022"],
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  },
  "exclude": ["node_modules", "cdk.out"]
}

{
  "app": "npx ts-node --prefer-ts-exts app.ts"
}

#!/usr/bin/env python3
"""VPCからBrave Search APIの「AI要約（Summarizer / AI Answers）」を使い、
質問に対する回答（要約＋引用）を取得するサンプル。

フロー（すべて api.search.brave.com の同一ドメインで完結）:
  1) /res/v1/web/search?q=...&summary=1   → レスポンスから summarizer.key を得る
  2) /res/v1/summarizer/search?key=...     → 要約（質問への回答）を取得する
要約・本文抽出はBraveのサーバー側で行われるため、外向きの穴は1ドメインのまま。
"""
import json
import sys
import urllib.parse
import urllib.request
import urllib.error

REGION = "ap-northeast-1"
SECRET_ID = "brave-search/api-key"
BASE = "https://api.search.brave.com/res/v1"

def get_api_key() -> str:
    """Secrets Manager からBrave APIキーを取得する（boto3 が無ければ AWS CLI 経由）。"""
    try:
        import boto3  # type: ignore

        return boto3.client("secretsmanager", region_name=REGION).get_secret_value(
            SecretId=SECRET_ID
        )["SecretString"]
    except ImportError:
        import subprocess

        return subprocess.run(
            ["aws", "secretsmanager", "get-secret-value", "--secret-id", SECRET_ID,
             "--query", "SecretString", "--output", "text", "--region", REGION],
            capture_output=True, text=True, check=True,
        ).stdout.strip()

def get_json(url: str, key: str):
    """GET して (HTTPステータス, JSON) を返す。HTTPErrorでも本文をJSONで返す。"""
    req = urllib.request.Request(
        url, headers={"Accept": "application/json", "X-Subscription-Token": key}
    )
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            return r.status, json.loads(r.read())
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read() or b"{}")
        except Exception:
            return e.code, {}

def extract_summary_text(d: dict) -> str:
    """要約レスポンスから本文テキストを取り出す。

    summary は {type, data} の配列で、data が文字列のもの（回答テキスト）と
    dict のもの（画像・エンティティ等）が混在する。文字列の data だけを連結する。
    """
    parts = []
    summ = d.get("summary")
    if isinstance(summ, list):
        for p in summ:
            if isinstance(p, str):
                parts.append(p)
            elif isinstance(p, dict):
                data = p.get("data")
                if isinstance(data, str):
                    parts.append(data)
                elif isinstance(data, dict):
                    t = data.get("text") or data.get("title")
                    if isinstance(t, str):
                        parts.append(t)
    return "".join(parts).strip()

def main() -> None:
    question = sys.argv[1] if len(sys.argv) > 1 else "What is AWS Network Firewall?"
    key = get_api_key()

    # 1) summary=1 付きで検索して summarizer key を得る
    url1 = f"{BASE}/web/search?" + urllib.parse.urlencode(
        {"q": question, "summary": 1, "count": 5}
    )
    st1, d1 = get_json(url1, key)
    print(f"[1] web/search?summary=1   http={st1}  question={question!r}")

    skey = (d1.get("summarizer") or {}).get("key")
    if not skey:
        print("    -> summarizerキーが返りませんでした（プラン未対応 or 要約対象外クエリ）。")
        print("    -> 代わりに通常の検索スニペットを表示します:")
        for r in (d1.get("web", {}).get("results", []) or [])[:3]:
            print("    -", r.get("title"))
            print("      ", (r.get("description") or "")[:160])
        return

    # 2) summarizer/search で回答（要約）を取得
    url2 = f"{BASE}/summarizer/search?" + urllib.parse.urlencode(
        {"key": skey, "inline_references": "true"}
    )
    st2, d2 = get_json(url2, key)
    print(f"[2] summarizer/search      http={st2}")

    text = extract_summary_text(d2)
    print("=== ANSWER (Braveからの要約回答) ===")
    print(text[:2000] if text else json.dumps(d2, ensure_ascii=False)[:1500])

if __name__ == "__main__":
    main()