apiVersion: v1
kind: ServiceAccount
metadata:
  name: s3-access-sa
  namespace: default

{
  "aud": [
    "https://koreacentral.oic.prod-aks.azure.com/00000000-0000-0000-0000-000000000000/11111111-1111-1111-1111-111111111111/",
    "https://<cluster-name>.hcp.koreacentral.azmk8s.io",
    "\"<cluster-name>.hcp.koreacentral.azmk8s.io\""
  ],
  "exp": 1755673306,
  "iat": 1755669706,
  "iss": "https://koreacentral.oic.prod-aks.azure.com/00000000-0000-0000-0000-000000000000/11111111-1111-1111-1111-111111111111/",
  "jti": "a8c8e96a-2aea-4a3f-8aa2-779b2bee8853",
  "kubernetes.io": {
    "namespace": "default",
    "serviceaccount": {
      "name": "s3-access-sa",
      "uid": "c400e143-64db-4338-a986-41001239c272"
    }
  },
  "nbf": 1755669706,
  "sub": "system:serviceaccount:default:s3-access-sa"
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::<AWS_ACCOUNT_ID>:oidc-provider/<OIDC_ISSUER_URL_WITHOUT_HTTPS>"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "StringEquals": {
          "<OIDC_ISSUER_URL_WITHOUT_HTTPS>:sub": "system:serviceaccount:default:s3-access-sa",
          "<OIDC_ISSUER_URL_WITHOUT_HTTPS>:aud": "https://<OIDC_ISSUER_URL_WITHOUT_HTTPS>/"
        }
      }
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
      "Resource": "arn:aws:s3:::<MY_S3_BUCKET>/*"
    },
    {
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::<MY_S3_BUCKET>"
    }
  ]
}

import boto3
import os
import logging
from datetime import datetime, timezone, timedelta

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 한국 표준시 (UTC+9)
KST = timezone(timedelta(hours=9))

def verify_aws_identity() -> bool:
    """현재 AWS 자격 증명 정보를 확인하고 출력합니다."""
    try:
        logger.info("현재 AWS 자격 증명을 확인합니다...")
        sts_client = boto3.client('sts')
        identity = sts_client.get_caller_identity()
        logger.info(f"  - AWS 계정: {identity['Account']}")
        logger.info(f"  - 역할 세션 ARN: {identity['Arn']}")
    except Exception as e:
        logger.error(f"❌ AWS 자격 증명 확인 실패: {e}")
        return False
    return True

def upload_to_s3() -> None:
    """boto3를 사용하여 S3에 파일을 업로드합니다."""
    local_file = 'aks-oidc-test.txt'
    try:
        # 환경변수에서 S3 버킷 이름과 리전 정보 가져오기
        bucket_name = os.getenv('S3_BUCKET_NAME')
        region = os.getenv('AWS_REGION', 'ap-northeast-2')

        if not bucket_name:
            raise ValueError("S3_BUCKET_NAME 환경변수가 설정되지 않았습니다.")

        logger.info(f"대상 S3 버킷: {bucket_name}, 리전: {region}")

        # boto3 클라이언트는 환경변수(AWS_ROLE_ARN, AWS_WEB_IDENTITY_TOKEN_FILE)를
        # 기반으로 자동으로 AssumeRoleWithWebIdentity를 호출합니다.
        s3_client = boto3.client('s3', region_name=region)

        # 테스트용 파일 내용 생성
        timestamp_str = datetime.now(KST).strftime('%Y-%m-%d %H:%M:%S KST')
        file_content = f"AKS OIDC S3 Upload Test\nTimestamp: {timestamp_str}\n"
        with open(local_file, 'w', encoding='utf-8') as f:
            f.write(file_content)

        # S3에 업로드할 객체 키 정의
        s3_key = f"uploads/aks-oidc-test-{datetime.now(KST).strftime('%Y%m%d_%H%M%S')}.txt"

        logger.info(f"파일 업로드를 시작합니다: {local_file} -> s3://{bucket_name}/{s3_key}")
        s3_client.upload_file(local_file, bucket_name, s3_key)

        logger.info(f"✅ 파일 업로드 성공! s3://{bucket_name}/{s3_key}")

    except Exception as e:
        logger.error(f"❌ 업로드 중 오류 발생: {e}", exc_info=True)
    finally:
        if os.path.exists(local_file):
            os.remove(local_file)

if __name__ == "__main__":
    logger.info("=== AKS to S3 OIDC 테스트 시작 ===")
    if verify_aws_identity():
        upload_to_s3()
    logger.info("=== 테스트 종료 ===")

apiVersion: batch/v1
kind: Job
metadata:
  name: aks-oidc-s3-test-job
  namespace: default
spec:
  template:
    spec:
      # 1. 앞에서 생성한 ServiceAccount 지정
      serviceAccountName: s3-access-sa
      containers:
      - name: s3-uploader
        image: "<YOUR_ACR_REGISTRY>.azurecr.io/aks-oidc-s3-demo:latest"
        env:
        # boto3가 자동으로 인식하는 환경변수 설정
        - name: S3_BUCKET_NAME
          value: "<MY_S3_BUCKET>"
        - name: AWS_REGION
          value: "ap-northeast-2"
        - name: AWS_ROLE_ARN
          value: "<MY_IAM_ROLE_ARN>"
        - name: AWS_WEB_IDENTITY_TOKEN_FILE
          value: "/var/run/secrets/tokens/aws-iam-token"
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "256Mi"
            cpu: "200m"
        volumeMounts:
        # 3. Projected Volume을 통해 생성된 토큰을 컨테이너에 마운트
        - name: aws-iam-token
          mountPath: "/var/run/secrets/tokens"
          readOnly: true
      volumes:
      # 2. Kubernetes API에 특정 Audience를 가진 OIDC 토큰을 요청하여 볼륨으로 생성
      - name: aws-iam-token
        projected:
          sources:
          - serviceAccountToken:
              path: aws-iam-token
              # audience는 AWS OIDC 공급자에 등록한 값과 정확히 일치해야 함
              # 기본값 사용 시 여러 audience 값이 배열로 들어가 AWS 인증 실패
              audience: "https://koreacentral.oic.prod-aks.azure.com/tenant-id/cluster-id/" # 반드시 끝에 "/"가 있어야 함
              expirationSeconds: 3600 # 1시간
      restartPolicy: Never
  backoffLimit: 3

boto3==1.34.144

FROM python:3.11-slim

WORKDIR /app

# 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 애플리케이션 코드 복사
COPY app/ .

# 애플리케이션 실행
CMD ["python", "main.py"]

# Docker 이미지 빌드
docker build -t aks-oidc-s3-demo .

# ACR 로그인
az acr login --name <YOUR_ACR_NAME>

# 이미지 태그 지정 및 푸시
docker tag aks-oidc-s3-demo <YOUR_ACR_NAME>.azurecr.io/aks-oidc-s3-demo:latest
docker push <YOUR_ACR_NAME>.azurecr.io/aks-oidc-s3-demo:latest

# 1. ServiceAccount 배포
kubectl apply -f k8s/serviceaccount.yaml

# 2. Job 배포 (job.yaml 파일의 플레이스홀더 값을 실제 값으로 변경했는지 확인!)
kubectl apply -f k8s/job.yaml

# 3. Job Pod 상태 확인
kubectl get pods --selector=job-name=aks-oidc-s3-test-job

# 4. Pod 로그 확인하여 S3 업로드 성공 여부 확인
POD_NAME=$(kubectl get pods --selector=job-name=aks-oidc-s3-test-job -o jsonpath='{.items[0].metadata.name}')
kubectl logs -f $POD_NAME