Amazon WorkSpaces Applicationsのデフォルトのユーザープールではメールアドレスとパスワードがあれば簡単にユーザーを作成することができます。
しかし、メールアドレスとパスワードだけでログインできる状況は昨今の状況を見るにセキュアではないので最低限MFAが使用できるようにしておきたいです。
https://docs.aws.amazon.com/ja_jp/appstream2/latest/developerguide/user-pool.html
ユーザープールでMFAを使うには以下のように記載されているため、外部のIdPを用意してあげる必要があります。
https://docs.aws.amazon.com/ja_jp/appstream2/latest/developerguide/client-features.html
WorkSpaces アプリケーションの場合、MFA は SAML 2.0 を介してサポートされています。
IdPを準備しないでもMFAくらいは使えるようにしたいのでCognitoのユーザープールを組み合わせて認証に成功した際にストリーミングURLを返すようにしてみます。
やること
Amazon WorkSpaces Applicationsの作成
こちらは前回作成した環境を再利用します。
Amazon Cognito、Amazon API Gateway、AWS Lambdaの作成
こちらはユーザーの管理とストリーミングURLの生成に使用します。
構成図にすると以下のようなイメージになります。
設定
CognitoとAPI Gateway + Lambda作成
WorkSpaces Applications部分は前回のブログと同じないようなので紹介は割愛します。
Cognito、API Gateway、LambdaはTerraformで作成します。
作成に使用するTerraformのコードは以下です。
Cognitoはマネージドログインを使用していてログイン画面などはこちらで作成しないでも使えるようにしています。
TerraformとLambdaのコードはCodexに作成してもらいました。
main.tf
data "aws_caller_identity" "current" {}
data "aws_partition" "current" {}
locals {
group_name = "streaming-users"
login_url = "https://${aws_cognito_user_pool_domain.portal.domain}.auth.${var.aws_region}.amazoncognito.com"
}
resource "aws_cognito_user_pool" "portal" {
name = var.name
user_pool_tier = "ESSENTIALS"
username_attributes = ["email"]
mfa_configuration = "ON"
admin_create_user_config {
allow_admin_create_user_only = true
}
software_token_mfa_configuration {
enabled = true
}
account_recovery_setting {
recovery_mechanism {
name = "admin_only"
priority = 1
}
}
password_policy {
minimum_length = 14
require_lowercase = true
require_uppercase = true
require_numbers = true
require_symbols = true
temporary_password_validity_days = 7
}
}
resource "aws_cognito_user_pool_domain" "portal" {
domain = var.cognito_domain_prefix
user_pool_id = aws_cognito_user_pool.portal.id
managed_login_version = 2
}
resource "aws_cognito_user_pool_client" "portal" {
name = var.name
user_pool_id = aws_cognito_user_pool.portal.id
generate_secret = true
allowed_oauth_flows_user_pool_client = true
allowed_oauth_flows = ["code"]
allowed_oauth_scopes = ["openid"]
supported_identity_providers = ["COGNITO"]
callback_urls = ["${aws_apigatewayv2_api.portal.api_endpoint}/callback"]
logout_urls = ["${aws_apigatewayv2_api.portal.api_endpoint}/"]
explicit_auth_flows = ["ALLOW_REFRESH_TOKEN_AUTH"]
prevent_user_existence_errors = "ENABLED"
enable_token_revocation = true
access_token_validity = 5
id_token_validity = 5
refresh_token_validity = 1
token_validity_units {
access_token = "minutes"
id_token = "minutes"
refresh_token = "days"
}
}
resource "aws_cognito_managed_login_branding" "portal" {
client_id = aws_cognito_user_pool_client.portal.id
user_pool_id = aws_cognito_user_pool.portal.id
use_cognito_provided_values = true
depends_on = [aws_cognito_user_pool_domain.portal]
}
resource "aws_cognito_user_group" "streaming" {
name = local.group_name
user_pool_id = aws_cognito_user_pool.portal.id
description = "Administrators grant streaming access by adding users."
}
resource "aws_apigatewayv2_api" "portal" {
name = var.name
protocol_type = "HTTP"
}
resource "aws_apigatewayv2_stage" "portal" {
api_id = aws_apigatewayv2_api.portal.id
name = "$default"
auto_deploy = true
default_route_settings {
throttling_burst_limit = 10
throttling_rate_limit = 5
}
}
resource "aws_apigatewayv2_integration" "portal" {
api_id = aws_apigatewayv2_api.portal.id
integration_type = "AWS_PROXY"
integration_uri = aws_lambda_function.portal.invoke_arn
payload_format_version = "2.0"
}
resource "aws_apigatewayv2_route" "public" {
for_each = toset(["GET /", "GET /callback"])
api_id = aws_apigatewayv2_api.portal.id
route_key = each.value
target = "integrations/${aws_apigatewayv2_integration.portal.id}"
}
resource "aws_iam_role" "portal" {
name = "${var.name}-lambda"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Principal = { Service = "lambda.amazonaws.com" }
Action = "sts:AssumeRole"
}]
})
}
resource "aws_cloudwatch_log_group" "portal" {
name = "/aws/lambda/${var.name}"
retention_in_days = 14
}
resource "aws_iam_role_policy" "portal" {
role = aws_iam_role.portal.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = ["logs:CreateLogStream", "logs:PutLogEvents"]
Resource = "${aws_cloudwatch_log_group.portal.arn}:*"
},
{
Effect = "Allow"
Action = ["appstream:CreateStreamingURL"]
Resource = [
"arn:${data.aws_partition.current.partition}:appstream:${var.aws_region}:${data.aws_caller_identity.current.account_id}:stack/${var.stack_name}",
"arn:${data.aws_partition.current.partition}:appstream:${var.aws_region}:${data.aws_caller_identity.current.account_id}:fleet/${var.fleet_name}"
]
},
{
Effect = "Allow"
Action = ["cognito-idp:AdminGetUser", "cognito-idp:AdminListGroupsForUser", "cognito-idp:AdminGetUserAuthFactors", "cognito-idp:GetUserPoolMfaConfig"]
Resource = aws_cognito_user_pool.portal.arn
}
]
})
}
data "archive_file" "portal" {
type = "zip"
source_dir = "${path.module}/lambda"
output_path = "${path.module}/portal.zip"
}
# Build with python3 scripts/build_sdk.py before terraform plan/apply.
# Pin the SDK because older Lambda runtime SDKs lack AdminGetUserAuthFactors.
resource "aws_lambda_layer_version" "sdk" {
layer_name = "${var.name}-sdk"
filename = "${path.module}/.build/cognito-sdk.zip"
source_code_hash = filebase64sha256("${path.module}/.build/cognito-sdk.zip")
compatible_runtimes = ["python3.13"]
}
resource "aws_lambda_function" "portal" {
function_name = var.name
role = aws_iam_role.portal.arn
runtime = "python3.13"
layers = [aws_lambda_layer_version.sdk.arn]
handler = "handler.handler"
filename = data.archive_file.portal.output_path
source_code_hash = data.archive_file.portal.output_base64sha256
timeout = 15
memory_size = 128
environment {
variables = {
POOL_ID = aws_cognito_user_pool.portal.id
CLIENT_ID = aws_cognito_user_pool_client.portal.id
CLIENT_SECRET = aws_cognito_user_pool_client.portal.client_secret
LOGIN_URL = local.login_url
CALLBACK_URL = "${aws_apigatewayv2_api.portal.api_endpoint}/callback"
GROUP_NAME = local.group_name
STACK_NAME = var.stack_name
FLEET_NAME = var.fleet_name
}
}
depends_on = [aws_iam_role_policy.portal]
}
resource "aws_lambda_permission" "gateway" {
statement_id = "AllowAPIGateway"
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.portal.function_name
principal = "apigateway.amazonaws.com"
source_arn = "${aws_apigatewayv2_api.portal.execution_arn}/*/*"
}
terraform.tfvars
aws_region = "ap-northeast-1"
name = "workspaces-mfa"
cognito_domain_prefix = "workspaces-mfa"
stack_name = "test-stack"
fleet_name = "test-fleet"
variables.tf
variable "aws_region" {
type = string
default = "ap-northeast-1"
}
variable "name" {
type = string
default = "workspaces-mfa"
validation {
condition = can(regex("^[a-z][a-z0-9-]{2,39}$", var.name))
error_message = "name must be 3-40 lowercase letters, digits or hyphens, starting with a letter."
}
}
variable "cognito_domain_prefix" {
type = string
description = "Unique Cognito domain prefix in the region."
}
variable "stack_name" {
type = string
description = "Existing WorkSpaces Applications stack in this account and region."
}
variable "fleet_name" {
type = string
description = "Existing fleet associated with the stack."
}
Lambdaのコードは以下のようになっています。
API Gatewayの「/」にGETでアクセスするとbegin_login()という関数が実行されCognitoのマネージドログインのURLに飛ばすようにしています。
その後、Cognitoでログインが成功したらAPI Gatewayの「/callback」にリクエストを送ります。
「/callback」にアクセスが発生しread_login()が実行されます。
read_login()ではCookieの内容を確認して通して問題ないかの判定を行っています。
その後、stream(sub)でCognitoのユーザー情報を確認してからストリーミングURLを返す動きになっています。
handler.py
import base64
import hashlib
import hmac
import json
import os
import secrets
import time
from http.cookies import CookieError, SimpleCookie
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from uuid import UUID
import boto3
from botocore.exceptions import BotoCoreError, ClientError
cognito = boto3.client("cognito-idp")
appstream = boto3.client("appstream")
COOKIE = "__Host-streaming-login"
TTL = 900
def b64(data):
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def signature(payload):
return b64(hmac.new(os.environ["CLIENT_SECRET"].encode(),
("login-cookie:" + payload).encode(), hashlib.sha256).digest())
def cookie(value="", max_age=0):
return f"{COOKIE}={value}; Path=/; Max-Age={max_age}; Secure; HttpOnly; SameSite=Lax"
def response(status, body="", location=None, login_cookie=None):
result = {
"statusCode": status,
"headers": {
"content-type": "text/plain; charset=utf-8",
"cache-control": "no-store",
"referrer-policy": "no-referrer",
"x-content-type-options": "nosniff",
"strict-transport-security": "max-age=31536000",
"content-security-policy": "default-src 'none'; frame-ancestors 'none'",
},
"body": body,
}
if location:
result["headers"]["location"] = location
if login_cookie is not None:
result["cookies"] = [login_cookie]
return result
def begin_login():
state, verifier = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
payload = b64(json.dumps({
"state": state, "verifier": verifier, "expires": int(time.time()) + TTL
}).encode())
params = urlencode({
"client_id": os.environ["CLIENT_ID"], "response_type": "code", "scope": "openid",
"redirect_uri": os.environ["CALLBACK_URL"], "state": state,
"code_challenge_method": "S256",
"code_challenge": b64(hashlib.sha256(verifier.encode()).digest()),
"prompt": "login",
})
return response(302, location=os.environ["LOGIN_URL"] + "/oauth2/authorize?" + params,
login_cookie=cookie(payload + "." + signature(payload), TTL))
def read_login(event, state):
jar = SimpleCookie()
jar.load("; ".join(event.get("cookies", [])))
value = jar[COOKIE].value
payload, mac = value.split(".")
if not hmac.compare_digest(mac, signature(payload)):
raise ValueError("Invalid cookie signature")
data = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
if (data["expires"] <= time.time()
or not state
or not hmac.compare_digest(data["state"], state)):
raise ValueError("Invalid state or expired cookie")
return data["verifier"]
def request_json(request):
with urlopen(request, timeout=5) as result:
return json.load(result)
def exchange_identity(code, verifier):
credentials = base64.b64encode(
(os.environ["CLIENT_ID"] + ":" + os.environ["CLIENT_SECRET"]).encode()
).decode()
tokens = request_json(Request(
os.environ["LOGIN_URL"] + "/oauth2/token",
data=urlencode({
"grant_type": "authorization_code", "code": code,
"redirect_uri": os.environ["CALLBACK_URL"], "code_verifier": verifier,
}).encode(),
headers={"Authorization": "Basic " + credentials,
"Content-Type": "application/x-www-form-urlencoded"},
method="POST",
))
# Tokens come only from our fixed Cognito HTTPS endpoint, never from the browser.
# userInfo validates the access token; do not trust an unverified JWT payload.
identity = request_json(Request(
os.environ["LOGIN_URL"] + "/oauth2/userInfo",
headers={"Authorization": "Bearer " + tokens["access_token"]},
))
return str(UUID(identity["sub"]))
def stream(sub):
args = {"UserPoolId": os.environ["POOL_ID"], "Username": sub}
user = cognito.admin_get_user(**args)
attributes = {a["Name"]: a["Value"] for a in user["UserAttributes"]}
if not user.get("Enabled"):
return response(403, "アカウントが無効です。管理者にお問い合わせください。[ACCOUNT_DISABLED]",
login_cookie=cookie())
if user.get("UserStatus") != "CONFIRMED":
return response(403, "ユーザー状態がCONFIRMEDではありません。管理者に状態の確認を依頼してください。[USER_NOT_CONFIRMED]",
login_cookie=cookie())
if attributes.get("sub") != sub:
return response(403, "認証したユーザーと取得したユーザー情報が一致しません。管理者にお問い合わせください。[IDENTITY_MISMATCH]",
login_cookie=cookie())
# Required MFA is enforced by Cognito during managed login. Recheck the live
# pool setting so changing it to OPTIONAL/OFF cannot silently weaken access.
mfa = cognito.get_user_pool_mfa_config(UserPoolId=os.environ["POOL_ID"])
if (mfa.get("MfaConfiguration") != "ON"
or mfa.get("SoftwareTokenMfaConfiguration", {}).get("Enabled") is not True):
return response(403, "ユーザープールのMFA必須・TOTP有効設定を確認できません。[POOL_MFA_NOT_REQUIRED]",
login_cookie=cookie())
args["Username"] = user["Username"]
factors = cognito.admin_get_user_auth_factors(**args)
if factors.get("Username") != user["Username"]:
return response(403, "認証方式の取得対象が一致しません。[IDENTITY_MISMATCH]",
login_cookie=cookie())
# UserMFASettingList can be absent even after successful TOTP sign-in.
# ConfiguredUserAuthFactors reports actual enrollment, including SOFTWARE_TOKEN.
if "SOFTWARE_TOKEN" not in (factors.get("ConfiguredUserAuthFactors") or []):
return response(403, "認証アプリのTOTP登録を確認できません。[TOTP_NOT_REGISTERED]",
login_cookie=cookie())
while True:
groups = cognito.admin_list_groups_for_user(**args)
if any(g["GroupName"] == os.environ["GROUP_NAME"] for g in groups["Groups"]):
break
if not groups.get("NextToken"):
return response(403, "管理者による利用許可が必要です。", login_cookie=cookie())
args["NextToken"] = groups["NextToken"]
result = appstream.create_streaming_url(
StackName=os.environ["STACK_NAME"], FleetName=os.environ["FLEET_NAME"],
UserId=UUID(sub).hex, Validity=60,
)
return response(302, location=result["StreamingURL"], login_cookie=cookie())
def handler(event, context):
route = event.get("routeKey")
if route == "GET /":
return begin_login()
if route != "GET /callback":
return response(404, "Not found")
try:
query = event.get("queryStringParameters") or {}
verifier = read_login(event, query.get("state"))
if query.get("error") or not query.get("code"):
return response(400, "認証が完了しませんでした。入口URLから再度ログインしてください。",
login_cookie=cookie())
return stream(exchange_identity(query["code"], verifier))
except HTTPError as error:
status = 400 if error.code < 500 else 503
return response(status, "認証を完了できません。入口URLから再度ログインしてください。",
login_cookie=cookie())
except (ValueError, KeyError, TypeError, CookieError):
return response(400, "認証リクエストが無効か期限切れです。入口URLから再度ログインしてください。",
login_cookie=cookie())
except (URLError, TimeoutError, BotoCoreError, ClientError):
# Never log complete events, authorization codes, cookies, tokens or URLs.
return response(503, "接続を開始できません。管理者にお問い合わせください。",
login_cookie=cookie())
Terrafomのデプロイは以下のコマンドで実施できます。
terraform init
terraform apply
Cognitoにユーザーを作る
AWSリソースの作成ができたらCognitoにユーザーを追加します。
Cognitoのコンソールからworkspaces-mfaというユーザープールを開いてください。
開いたら画面左の項目からユーザーをクリックしてください。
クリックすると以下の画面が開きます。
そこからユーザーを作成をクリックします。
クリックしたらEメールアドレスと初期パスワードを入力してください。
パスワードは初回ログイン時に変更します。
ユーザーが作成できたらグループに追加します。
作成したユーザーをクリックして画面下にあるグループメンバーシップ欄からstreaming-usersというグループに追加してください。
ここまで出来たら準備完了です。
ログインを試してみる
準備ができたらAPI Gatewayのドメインにブラウザからアクセスします。(API Gatewayのコンソールから確認できます)
https://hogehoge.execute-api.ap-northeast-1.amazonaws.com/
アクセスすると以下のようにCognitoのマネージドログインの画面が表示されます。
この画面で先ほど作成したユーザーのメールアドレスとパスワードを入力します。
新しいパスワードの設定画面が出てくるのでルールに沿ってパスワードを設定します。
次の画面でMFAの設定を実施します。
Google Authenticatorや1Passwordなどがあれば設定可能です。
設定が完了したら以下のようにWorkSpaces Applicationの画面が出てくるのでここまでくれば成功です。
さいごに
今回はCognitoを使ってAmazon WorkSpaces Applicationsへアクセスするユーザーの管理を実施してみました。
IdPが用意できるならそちらを使った方がよいのですが、すぐに準備できないケースや外部のユーザー用に用意したいというケースもあると思うので、こういった回避策は検討対象になるのではないかと思います。