おい、ちょっと手を止めてくれ。昨日、あるステージング環境のインシデント調査を手伝ったんだがね、まあ見事にやられていたよ。何があったと思う?
KubernetesのAPIサーバーに、インターネット側から丸見えの状態で「匿名リクエスト(Anonymous Request)」が通る設定になっていたんだ。攻撃者はそれを足がかりにしてクラスタ内の情報を列挙し、最終的にデプロイされていたコンテナの機密環境変数を引っこ抜いていきやがった。
「えっ、うちはちゃんと社内ネットワークからしかアクセスできないようにファイアウォールを絞ってるから大丈夫だよ」……なんて甘いことを言ったそこの君。クラスタ内のポッドが踏み台にされたり、クラウドのセキュリティグループのちょっとしたミスで穴が空いた瞬間、Kubernetesのデフォルト設定の牙が君の喉元に食い込むことになる。
今日は、Kubernetes APIサーバーの「匿名認証の無効化」と、実務で必須になる「外部認証プロキシ(OIDC / Webhook)の構成」について、現場の泥臭い実態を踏まえて徹底的に解説する。教科書通りの綺麗な説明はおしまいだ。今日から君のインフラをガチガチに要塞化するための実践的な話をしよう。
—
1. なぜ「匿名認証(Anonymous Authentication)」が致命傷になるのか
KubernetesのAPIサーバーは、デフォルトで認証ヘッダーを持たないリクエストを「匿名ユーザー(system:anonymous)」として受け入れる設定になっている場合がある。
「いや、匿名ユーザーには権限を与えていないから安全だ」と思うかもしれない。しかし、セキュリティの基本原則を思い出してほしい。「攻撃者に無駄な情報を与えるな」だ。
匿名アクセスを許可していると、認証なしで以下のような情報が丸見えになる。
- クラスタのバージョン情報やAPIのエンドポイント一覧
- 権限昇格(Privilege Escalation)の糸口になり得るリソースの存在確認
- 誤って
ClusterRoleがバインドされたサービスアカウントの悪用可能性
まずは、この「入り口のドアの鍵を開けっぱなしにする仕様」を根本から断ち切る必要がある。
—
2. APIサーバーの匿名認証を完全に無効化する設定
Kubernetes APIサーバーの起動オプション、あるいはマネージドサービス(EKS, GKE, AKSなど)の設定で、匿名認証を明示的に禁止する。
自前で構築したK8s(kubeadm等)であれば、APIサーバーの静的ポッドマニフェスト(/etc/kubernetes/manifests/kube-apiserver.yaml)を次のように修正する。
apiVersion: v1
kind: Pod
metadata:
name: kube-apiserver
namespace: kube-system
spec:
containers:
- name: kube-apiserver
image: k8s.gcr.io/kube-apiserver:v1.28.0
command:
- kube-apiserver
# 以下のフラグを追加して、匿名認証を確実にシャットアウトする
- --anonymous-auth=false
# 注意: これを有効にすると、ヘルスチェックや一部の内部プローブが弾かれることがあるため、
# 適切なサービスアカウント認証が構成されていることを確認すること。
この --anonymous-auth=false を入れるだけで、認証トークンを持たないリクエストは容赦なく 401 Unauthorized で弾かれるようになる。まずはここがスタートラインだ。
—
3. 認証プロキシ(OAuth2 / OIDC)による厳格なアクセス制御の実装
匿名認証を切っただけでは、正当なユーザーや外部システムがAPIを叩けない。そこで、フロントに認証プロキシを置き、厳格なトークン検証を行った上でAPIサーバーへリクエストを中継するアーキテクチャが必要になる。
今回は、現場でよく使われる Nginx + 外部認証サービス(OAuth2 Proxy等) を模した、APIリクエストを安全に検証・転送するカスタム認証プロキシ(Python製)のサンプルコードを提示しよう。
このスクリプトは、クライアントからのリクエストに含まれるBearerトークンを検証し、ホワイトリスト化されたロールを持つユーザーのみをKubernetes APIサーバーへ通す仕組みだ。
セキュアな認証プロキシの実装サンプル(Python / FastAPI)
import os
import requests
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.responses import JSONResponse
app = FastAPI()
# 転送先のKubernetes APIサーバーの内部エンドポイント
K8S_API_SERVER = os.getenv("K8S_API_SERVER", "https://kubernetes.default.svc:443")
# 検証用のOIDC / 外部IDPのエンドポイント(例: KeycloakやAuth0)
OIDC_INTROSPECT_URL = os.getenv("OIDC_INTROSPECT_URL", "https://auth.example.com/oauth/check_token")
@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def proxy(request: Request, path: str):
# 1. リクエストヘッダーからAuthorizationトークンを取得
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
return JSONResponse(
status_code=401,
content={"error": "Missing or invalid Authorization header"}
)
token = auth_header.split(" ")[1]
# 2. 外部IDP(OIDC等)にトークンの正当性を検証(Introspection)
try:
# 実際の現場ではここでJWTの署名検証をローカル(JWKS)で行うのが高速でベスト
introspection_response = requests.post(
OIDC_INTROSPECT_URL,
data={"token": token},
timeout=5.0
)
if introspection_response.status_code != 200:
raise HTTPException(status_code=401, detail="Token validation failed")
token_data = introspection_response.json()
if not token_data.get("active", False):
return JSONResponse(status_code=401, content={"error": "Inactive token"})
# 3. 認可チェック(例: 特定のグループに属しているか)
user_groups = token_data.get("groups", [])
if "k8s-admins" not in user_groups:
return JSONResponse(status_code=403, content={"error": "Forbidden: Insufficient privileges"})
except requests.RequestException:
return JSONResponse(
status_code=503,
content={"error": "Authentication service unavailable"}
)
# 4. 検証成功:Kubernetes APIサーバーへリクエストを転送
# 本番環境では、mTLS(相互TLS)を用いてAPIサーバーと安全に通信すること
target_url = f"{K8S_API_SERVER}/{path}"
headers = dict(request.headers)
headers["Host"] = request.url.hostname
# クライアントの元IPを引き継ぐ
headers["X-Forwarded-For"] = request.client.host
body = await request.body()
try:
resp = requests.request(
method=request.method,
url=target_url,
headers=headers,
data=body,
verify="/var/run/secrets/kubernetes.io/serviceaccount/ca.crt", # K8sのCA証明書で検証
timeout=30.0
)
return Response(
content=resp.content,
status_code=resp.status_code,
headers=dict(resp.headers)
)
except requests.RequestException as e:
return JSONResponse(
status_code=502,
content={"error": f"Failed to connect to Kubernetes API: {str(e)}"}
)
if __name__ == "__main__":
import uvicorn
# セキュアにコンテナ内等で動作させる
uvicorn.run(app, host="0.0.0.0", port=8443, ssl_keyfile="/certs/server.key", ssl_certfile="/certs/server.crt")
—
4. 現場のチーフからの実践的なアドバイス
上記のプロキシや設定を入れただけでは、まだ油断してはいけない。実務の現場でインシデントを防ぐために、以下のチェックリストを必ずチームで回してほしい。
1. mTLS(相互TLS)の徹底
認証プロキシからAPIサーバーへの通信は、必ずクライアント証明書(mTLS)を用いた強固な経路暗号化と証明書検証を行うこと。平文の内部ネットワークであっても、万が一の侵入を想定してゼロトラストの思想で組むべきだ。
2. 監査ログ(Audit Logs)の有効化
誰が、いつ、どのAPIを叩いたか。匿名アクセスが無効化されている環境であれば、すべてのリクエストにアイデンティティが紐づく。Kubernetesの監査ポリシーを厳しく設定し、怪しい挙動(例えば、secrets リソースへの異常な大量アクセスなど)をSIEMやログ監視ツールでリアルタイム検知できるようにしておこう。
3. 定期的な脆弱性スキャン
自分たちがデプロイしたマニフェストに、誤って system:anonymous や過剰な権限(cluster-admin)を付与したバインドが残っていないか、kube-bench などのツールを使って定期的に自動監査を回す仕組みをCI/CDに組み込むこと。
セキュリティは「一度設定したら終わり」の静的なものではない。攻撃者は常に新しい隙を狙っている。だからこそ、こうした基本中の基本である「不要なアクセスの遮断と厳格な認証の強制」を泥臭くやり続けることが、最終的に会社と自分自身のキャリアを守る盾になるんだ。
さて、理屈はここまでだ。さっそく今日の業務で、自分たちのクラスタのAPIサーバーの設定を確認しに行こうか。
コメント