AI 时代如何快速搭建可视化数仓：使用 Amazon Quick App 快速搭建 Amazon Redshift 应用

export AWS_REGION=<REGION>
printf '%s\n' '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | uvx awslabs.redshift-mcp-server@latest

# 独立 venv 安装 mcp-proxy，并固定 mcp 到 1.x 兼容版本
uv venv /home/ec2-user/mcp-proxy-env --python 3.12
source /home/ec2-user/mcp-proxy-env/bin/activate
uv pip install mcp-proxy
uv pip install "mcp>=1.17.0,<2"

# 启动桥接：监听 8000，把请求转给 uvx 拉起的 MCP Server
export AWS_REGION=<REGION>
mcp-proxy --host 0.0.0.0 --port 8000 --pass-environment \
  -- uvx awslabs.redshift-mcp-server@latest

curl -s -X POST http://127.0.0.1:8000/mcp \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"1.0"}}}'

# Target Group（TCP/instance）
aws elbv2 create-target-group --name mcp-proxy-tg --protocol TCP --port 8000 \
  --vpc-id <VPC_ID> --target-type instance \
  --health-check-protocol TCP --health-check-port 8000 --region <REGION>
# 注册 EC2
aws elbv2 register-targets --target-group-arn <TG_ARN> \
  --targets Id=<EC2_INSTANCE_ID>,Port=8000 --region <REGION>
# 内部 NLB
aws elbv2 create-load-balancer --name mcp-proxy-nlb --type network \
  --scheme internal --subnets <SUBNET_ID> --region <REGION>
# 监听器 TCP 8000 -> Target Group
aws elbv2 create-listener --load-balancer-arn <NLB_ARN> --protocol TCP --port 8000 \
  --default-actions Type=forward,TargetGroupArn=<TG_ARN> --region <REGION>

aws ec2 authorize-security-group-ingress --group-id <SECURITY_GROUP_ID> \
  --ip-permissions 'IpProtocol=tcp,FromPort=8000,ToPort=8000,IpRanges=[{CidrIp=<VPC_CIDR>}]' \
  --region <REGION>

# 导入 REST API（OpenAPI 中 responseTransferMode=STREAM, timeoutInMillis=900000）
aws apigateway import-rest-api --body 'fileb://mcp-apigw.yaml' \
  --parameters endpointConfigurationTypes=REGIONAL --region <REGION>
# 创建 VPC Link 指向 NLB（等待 status=AVAILABLE，约数分钟）
aws apigateway create-vpc-link --name mcp-proxy-vpclink \
  --target-arns <NLB_ARN> --region <REGION>
# 将 {proxy+} ANY 方法的集成改为 VPC_LINK，并指向 NLB DNS
aws apigateway update-integration --rest-api-id <ApiId> \
  --resource-id <ResourceId> --http-method ANY --patch-operations \
    op=replace,path=/connectionType,value=VPC_LINK \
    op=replace,path=/connectionId,value=<VpcLinkId> \
    op=replace,path=/uri,value=http://<NLB_DNS>:8000/{proxy} \
  --region <REGION>
# 部署
aws apigateway create-deployment --rest-api-id <ApiId> --stage-name prod --region <REGION>

# 用 client_credentials 换取 access_token
curl -s -X POST "https://<domain>.auth.<REGION>.amazoncognito.com/oauth2/token" \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  -u '<CLIENT_ID>:<CLIENT_SECRET>' \
  -d 'grant_type=client_credentials&scope=<RESOURCE_SERVER>/invoke'
# 预期返回 access_token（JWT, token_use=access, alg=RS256）

"""
Lambda Authorizer for Amazon Quick MCP -> Redshift.
校验 Cognito 2LO (client_credentials) 签发的 access token。
纯标准库实现 RS256 验签，无第三方依赖。
"""
import json
import os
import time
import base64
import hashlib
import urllib.request

# ---- 环境变量（部署时注入）----
ISSUER = os.environ["COGNITO_ISSUER"]           # https://cognito-idp.<region>.amazonaws.com/<pool>
JWKS_URL = os.environ["JWKS_URL"]
EXPECTED_SCOPE = os.environ.get("EXPECTED_SCOPE", "")   # 可空；非空则要求 token scope 含此值
EXPECTED_CLIENT_ID = os.environ.get("EXPECTED_CLIENT_ID", "")  # 可空；非空则要求匹配

_JWKS_CACHE = {"keys": None, "ts": 0}
_JWKS_TTL = 3600


def _b64url_decode(data: str) -> bytes:
    data += "=" * (-len(data) % 4)
    return base64.urlsafe_b64decode(data)


def _get_jwks():
    now = time.time()
    if _JWKS_CACHE["keys"] and now - _JWKS_CACHE["ts"] < _JWKS_TTL:
        return _JWKS_CACHE["keys"]
    with urllib.request.urlopen(JWKS_URL, timeout=5) as r:
        keys = json.loads(r.read())["keys"]
    _JWKS_CACHE["keys"] = keys
    _JWKS_CACHE["ts"] = now
    return keys


def _int_from_bytes(b: bytes) -> int:
    return int.from_bytes(b, "big")


def _rsa_verify_rs256(message: bytes, signature: bytes, n: int, e: int) -> bool:
    """纯 Python RS256 (RSASSA-PKCS1-v1_5, SHA-256) 验签。"""
    k = (n.bit_length() + 7) // 8
    if len(signature) != k:
        return False
    s = _int_from_bytes(signature)
    if s >= n:
        return False
    # RSA 公钥运算: m = s^e mod n
    m = pow(s, e, n)
    em = m.to_bytes(k, "big")
    # 构造期望的 EMSA-PKCS1-v1_5 编码
    digest = hashlib.sha256(message).digest()
    # SHA-256 的 DER 前缀
    der_prefix = bytes([
        0x30, 0x31, 0x30, 0x0d, 0x06, 0x09, 0x60, 0x86,
        0x48, 0x01, 0x65, 0x03, 0x04, 0x02, 0x01, 0x05,
        0x00, 0x04, 0x20,
    ])
    t = der_prefix + digest
    ps_len = k - len(t) - 3
    if ps_len < 8:
        return False
    expected = b"\x00\x01" + b"\xff" * ps_len + b"\x00" + t
    # 常量时间比较
    return _consteq(em, expected)


def _consteq(a: bytes, b: bytes) -> bool:
    if len(a) != len(b):
        return False
    r = 0
    for x, y in zip(a, b):
        r |= x ^ y
    return r == 0


def _verify_token(token: str) -> dict:
    parts = token.split(".")
    if len(parts) != 3:
        raise ValueError("malformed token")
    header = json.loads(_b64url_decode(parts[0]))
    payload = json.loads(_b64url_decode(parts[1]))
    signature = _b64url_decode(parts[2])

    if header.get("alg") != "RS256":
        raise ValueError("unexpected alg")
    kid = header.get("kid")

    # 找到对应公钥
    jwk = None
    for k in _get_jwks():
        if k.get("kid") == kid:
            jwk = k
            break
    if not jwk:
        # 刷新缓存重试一次
        _JWKS_CACHE["ts"] = 0
        for k in _get_jwks():
            if k.get("kid") == kid:
                jwk = k
                break
    if not jwk:
        raise ValueError("signing key not found")

    n = _int_from_bytes(_b64url_decode(jwk["n"]))
    e = _int_from_bytes(_b64url_decode(jwk["e"]))
    signing_input = (parts[0] + "." + parts[1]).encode("ascii")
    if not _rsa_verify_rs256(signing_input, signature, n, e):
        raise ValueError("signature verification failed")

    # 校验 claims
    now = int(time.time())
    if payload.get("iss") != ISSUER:
        raise ValueError("issuer mismatch")
    if payload.get("exp", 0) < now:
        raise ValueError("token expired")
    if payload.get("token_use") != "access":
        raise ValueError("not an access token")
    if EXPECTED_CLIENT_ID and payload.get("client_id") != EXPECTED_CLIENT_ID:
        raise ValueError("client_id mismatch")
    if EXPECTED_SCOPE:
        scopes = (payload.get("scope") or "").split()
        if EXPECTED_SCOPE not in scopes:
            raise ValueError("insufficient scope")
    return payload


def _policy(principal: str, effect: str, resource: str, context=None):
    doc = {
        "principalId": principal,
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [{
                "Action": "execute-api:Invoke",
                "Effect": effect,
                # 用 * 覆盖整个 API，避免缓存导致的资源不匹配
                "Resource": resource,
            }],
        },
    }
    if context:
        doc["context"] = context
    return doc


def handler(event, context):
    # TOKEN 类型 authorizer: event['authorizationToken'] = "Bearer xxx"
    token_header = event.get("authorizationToken", "") or ""
    method_arn = event.get("methodArn", "*")
    # 允许资源范围：同一 API/stage 下所有方法
    # methodArn 形如 arn:aws:execute-api:region:acct:apiId/stage/METHOD/path
    try:
        arn_parts = method_arn.split(":")
        api_part = arn_parts[5].split("/")
        resource_glob = ":".join(arn_parts[:5]) + ":" + api_part[0] + "/" + api_part[1] + "/*"
    except Exception:
        resource_glob = method_arn

    if not token_header.lower().startswith("bearer "):
        # 无 bearer -> 触发 401（Unauthorized）
        raise Exception("Unauthorized")

    token = token_header[7:].strip()
    try:
        claims = _verify_token(token)
    except Exception as ex:
        print("token verify failed:", str(ex))
        # 校验失败 -> Deny (返回 403) ；无效/过期 token
        return _policy("client", "Deny", resource_glob)

    ctx = {
        "client_id": str(claims.get("client_id", "")),
        "scope": str(claims.get("scope", "")),
    }
    return _policy(str(claims.get("client_id", "client")), "Allow", resource_glob, ctx)

# 创建执行角色
aws iam create-role --role-name mcp-authorizer-role \
  --assume-role-policy-document file://trust.json --region <REGION>
aws iam attach-role-policy --role-name mcp-authorizer-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
# 打包并创建函数
zip -q authorizer.zip lambda_function.py
aws lambda create-function --function-name mcp-oauth-authorizer \
  --runtime python3.12 --handler lambda_function.handler --timeout 10 \
  --role arn:aws:iam::<ACCOUNT_ID>:role/mcp-authorizer-role \
  --zip-file fileb://authorizer.zip --region <REGION> \
  --environment 'Variables={COGNITO_ISSUER=https://cognito-idp.<REGION>.amazonaws.com/<USER_POOL_ID>,\
    JWKS_URL=https://cognito-idp.<REGION>.amazonaws.com/<USER_POOL_ID>/.well-known/jwks.json,\
    EXPECTED_SCOPE=<RESOURCE_SERVER>/invoke,EXPECTED_CLIENT_ID=<CLIENT_ID>}'

# 创建 TOKEN Authorizer
aws apigateway create-authorizer --rest-api-id <ApiId> \
  --name cognito-jwt-authorizer --type TOKEN \
  --authorizer-uri arn:aws:apigateway:<REGION>:lambda:path/2015-03-31/functions/\
arn:aws:lambda:<REGION>:<ACCOUNT_ID>:function:mcp-oauth-authorizer/invocations \
  --identity-source method.request.header.Authorization \
  --authorizer-result-ttl-in-seconds 0 --region <REGION>
# 授予 API Gateway 调用该 Lambda 的权限
aws lambda add-permission --function-name mcp-oauth-authorizer \
  --statement-id apigw --action lambda:InvokeFunction \
  --principal apigateway.amazonaws.com \
  --source-arn arn:aws:execute-api:<REGION>:<ACCOUNT_ID>:<ApiId>/authorizers/<AuthorizerId> \
  --region <REGION>
# 挂到 {proxy+} ANY 方法
aws apigateway update-method --rest-api-id <ApiId> --resource-id <ResourceId> \
  --http-method ANY --patch-operations \
    op=replace,path=/authorizationType,value=CUSTOM \
    op=replace,path=/authorizerId,value=<AuthorizerId> --region <REGION>
# 401 响应携带 WWW-Authenticate（指向元数据 URL）
aws apigateway put-gateway-response --rest-api-id <ApiId> \
  --response-type UNAUTHORIZED --status-code 401 \
  --response-parameters '{"gatewayresponse.header.WWW-Authenticate":\
    "'\''Bearer resource_metadata=\"https://<ApiId>.execute-api.<REGION>.amazonaws.com/prod/.well-known/oauth-protected-resource\"'\''"}' \
  --region <REGION>
# 重新部署
aws apigateway create-deployment --rest-api-id <ApiId> --stage-name prod --region <REGION>

BASE="https://<ApiId>.execute-api.<REGION>.amazonaws.com/prod"; MCP="$BASE/mcp"
# 1) 元数据路由（无需认证）-> 200 + JSON
curl -s "$BASE/.well-known/oauth-protected-resource"
# 2) 无 token -> 401 + WWW-Authenticate
curl -s -i -X POST "$MCP" -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}' | grep -iE '^HTTP|www-authenticate'
# 3) 无效 token -> 403
curl -s -o /dev/null -w '%{http_code}\n' -X POST "$MCP" \
  -H 'Authorization: Bearer bad.token' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}'
# 4) 有效 token -> 200 + MCP 握手
curl -s -X POST "$MCP" -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"t","version":"1.0"}}}'

aws sts get-caller-identity

{
    "UserId": "userId",
    "Account": "1234567890",
    "Arn": "arn:aws:iam::1234567890:user/<用户名>"
}

-- 由 redshift admin 在 redshift 编辑器中执行（IAM 用户名含冒号，需加双引号）
GRANT USAGE ON SCHEMA <schema 名称>  TO "IAM:<用户名>";
GRANT SELECT ON ALL TABLES IN SCHEMA <schema 名称> TO "IAM:<用户名>";
ALTER DEFAULT PRIVILEGES IN SCHEMA <schema 名称> GRANT SELECT ON TABLES TO "IAM:<用户名>";