Great question. Assuming you’re on Prefect 3.x, the best way to enforce per-flow/deployment AWS access on EKS is to run each flow pod with its own Kubernetes ServiceAccount that’s bound to a minimal IAM role via IRSA. You set that ServiceAccount per deployment using Kubernetes worker job variables. Recommended approach - Keep the Prefect Kubernetes worker’s own ServiceAccount limited to just creating/monitoring Jobs. - Create one ServiceAccount per access profile, each annotated with the corresponding IRSA role. - In each deployment, set

service_account_name

in the work pool job variables so the flow pod runs with the least-privileged role it needs. Why this works - IRSA uses a ServiceAccount annotation to map to an IAM role. The Prefect Kubernetes worker will create your flow run Job/Pod using the ServiceAccount you specify per deployment. - This cleanly separates the worker’s permissions from each flow’s permissions. Implementation steps 1) In AWS/IAM: Create roles with minimal policies (one per access profile). 2) In Kubernetes: Create ServiceAccounts annotated for IRSA, for example:

apiVersion: v1
kind: ServiceAccount
metadata:
  name: data-readonly-sa
  namespace: your-ns
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam::123456789012:role/data-readonly-role
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: s3-admin-sa
  namespace: your-ns
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam::123456789012:role/s3-admin-role

3) In Prefect deployments, set

service_account_name

per deployment: - Using prefect.yaml

deployments:
- name: "etl-readonly"
  entrypoint: "flows/etl.py:flow"
  work_pool:
    name: "kubernetes"
    job_variables:
      namespace: "your-ns"
      service_account_name: "data-readonly-sa"

- name: "s3-maintenance"
  entrypoint: "flows/maintenance.py:flow"
  work_pool:
    name: "kubernetes"
    job_variables:
      namespace: "your-ns"
      service_account_name: "s3-admin-sa"

- Using Python

from prefect import flow

flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="flows/etl.py:flow",
).deploy(
    name="etl-readonly",
    work_pool_name="kubernetes",
    job_variables={
        "namespace": "your-ns",
        "service_account_name": "data-readonly-sa",
    },
)

Notes and options - You can run a single Kubernetes work pool and set different

service_account_name

per deployment (recommended), or create multiple work pools with different base job templates if you want stronger defaults per team. - The

annotations

job variable applies to both Job and Pod metadata, but IRSA role mapping is taken from the ServiceAccount annotation, not a Pod annotation. - For governance, use Prefect Cloud RBAC to restrict who can modify deployments or work pools so users can’t elevate privileges by changing job variables. Doc links and references - Kubernetes deployment guide - Customize job variables - Customize base job templates - Work pools concepts - Kubernetes worker integration - Source: Kubernetes worker implementation (job variables and template behavior) If you share your current work pool/deployment config, I can help map each flow to the right ServiceAccount/IAM role.

Got it. The clean pattern on EKS is: one minimal IAM role per “permission set,” one Kubernetes ServiceAccount per role (annotated for IRSA), and each Prefect deployment points to the right ServiceAccount via

service_account_name

. That gives you least-privilege per flow. Example layout for your cases - flow1 needs S3 + EMR - flow2 needs S3 + Athena Do this: 1) Create IAM policies - S3 (scoped to a bucket/prefix):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:ListBucket"
      ],
      "Resource": "arn:aws:s3:::my-bucket",
      "Condition": {
        "StringLike": {
          "s3:prefix": [
            "projectA/*"
          ]
        }
      }
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:DeleteObject"
      ],
      "Resource": "arn:aws:s3:::my-bucket/projectA/*"
    }
  ]
}

- EMR (pick the flavor you use; examples): - EMR on EC2 (classic): minimally

emr:RunJobFlow

,

emr:AddJobFlowSteps

,

emr:Describe*

,

emr:List*

. You may also need

iam:PassRole

for the EMR service role if your flow creates clusters. - EMR Serverless:

emr-serverless:StartJobRun

,

GetJobRun

,

ListJobRuns

, plus access to the application ARN(s). - EMR on EKS:

emr-containers:StartJobRun

,

DescribeJobRun

,

ListJobRuns

on target virtual cluster and

iam:PassRole

for the job runtime role if required. - Athena (scope to your workgroup and output location):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "athena:StartQueryExecution",
        "athena:GetQueryExecution",
        "athena:GetQueryResults"
      ],
      "Resource": "arn:aws:athena:REGION:ACCOUNT:workgroup/analytics-wg"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::athena-query-results-bucket",
        "arn:aws:s3:::athena-query-results-bucket/prefix/*"
      ]
    }
  ]
}

2) Create one IAM role per flow (or per permission set) and restrict it to a specific ServiceAccount via IRSA trust policy - Trust relationship (update OIDC provider, namespace, and SA names):

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::123456789012:oidc-provider/oidc.eks.REGION.amazonaws.com/id/OIDC_ID"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "StringEquals": {
          "<http://oidc.eks.REGION.amazonaws.com/id/OIDC_ID:aud|oidc.eks.REGION.amazonaws.com/id/OIDC_ID:aud>": "<http://sts.amazonaws.com|sts.amazonaws.com>",
          "<http://oidc.eks.REGION.amazonaws.com/id/OIDC_ID:sub|oidc.eks.REGION.amazonaws.com/id/OIDC_ID:sub>": "system:serviceaccount:your-ns:prefect-flow1-sa"
        }
      }
    }
  ]
}

- Attach S3+EMR policy to role for flow1; S3+Athena policy to role for flow2. 3) Create Kubernetes ServiceAccounts annotated with the role ARNs

apiVersion: v1
kind: ServiceAccount
metadata:
  name: prefect-flow1-sa
  namespace: your-ns
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam::123456789012:role/prefect-flow1-role
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: prefect-flow2-sa
  namespace: your-ns
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam::123456789012:role/prefect-flow2-role

4) Point each Prefect deployment at the correct ServiceAccount - In prefect.yaml:

deployments:
- name: flow1-s3-emr
  entrypoint: "flows/flow1.py:flow"
  work_pool:
    name: "kubernetes"
    job_variables:
      namespace: "your-ns"
      service_account_name: "prefect-flow1-sa"

- name: flow2-s3-athena
  entrypoint: "flows/flow2.py:flow"
  work_pool:
    name: "kubernetes"
    job_variables:
      namespace: "your-ns"
      service_account_name: "prefect-flow2-sa"

- Or in Python: ``` flow1.from_source( source="https://github.com/your-org/your-repo", entrypoint="flows/flow1.py:flow", ).deploy( name="flow1-s3-emr", work_pool_name="kubernetes", job_variables={"namespace": "your-ns", "service_account_name": "prefect-flow1-sa"}, ) flow2.from_source(

source="https://github.com/your-org/your-repo", entrypoint="flows/flow2.py:flow", ).deploy( name="flow2-s3-athena", work_pool_name="kubernetes", job_variables={"namespace": "your-ns", "service_account_name": "prefect-flow2-sa"}, ) ``` Security and governance tips - Keep the Kubernetes worker’s own ServiceAccount minimal; it just needs to create/watch Jobs, not access AWS data. - Lock roles to specific service accounts with the

sub

condition in the trust policy (as shown) to prevent privilege escalation. - In Prefect Cloud, use RBAC to restrict who can edit deployments/work pools so users can’t change

service_account_name

. - Optionally separate by namespace and reflect that in both the trust policy and deployment

namespace

. Docs and references - Prefect: Kubernetes deployment guide, Customize job variables, Customize base job templates, Work pools concepts, Kubernetes worker integration - AWS: EKS IRSA (IAM roles for service accounts) Quick check - Are you using EMR on EC2, EMR Serverless, or EMR on EKS for flow1? I can tailor the exact IAM actions and a minimal example once I know the EMR flavor and the scope of your S3/Athena resources (bucket prefixes, workgroups).

Absolutely — here’s a concise, copy/paste-friendly set of snippets you can use to implement per-deployment permissions with IRSA on EKS and Prefect 3.x. What you’ll set up - One IAM role per permission set (flow1: S3+EMR, flow2: S3+Athena) - One Kubernetes ServiceAccount per role, annotated for IRSA - Each Prefect deployment points to the right ServiceAccount via

service_account_name

1) Terraform: IAM roles and policies for IRSA Replace placeholders (ACCOUNT_ID, REGION, bucket names/prefixes, app/workgroup ARNs, cluster name, namespace, SA names). ``` # --------- Inputs --------- variable "cluster_name" { type = string } variable "namespace" { type = string } variable "sa_flow1" { type = string } # e.g., "prefect-flow1-sa" variable "sa_flow2" { type = string } # e.g., "prefect-flow2-sa" # S3 and Athena resources variable "project_bucket" { type = string } # e.g., "my-bucket" variable "project_prefix" { type = string } # e.g., "projectA/*" variable "athena_results_bucket" { type = string } # e.g., "athena-results-bucket" variable "athena_results_prefix" { type = string } # e.g., "teamA/*" variable "athena_workgroup_arn" { type = string } # e.g., "arnawsathenaREGIONACCOUNT_ID:workgroup/analytics-wg" # EMR resources (pick your flavor) variable "emr_serverless_app_arn" { type = string } # e.g., "arnawsemr-serverlessREGIONACCOUNT_ID:/applications/APP_ID" # If using EMR on EC2 instead, leave serverless var empty and see alt policy below. data "aws_caller_identity" "current" {} data "aws_eks_cluster" "this" { name = var.cluster_name } locals { oidc_provider_url = data.aws_eks_cluster.this.identity[0].oidc[0].issuer oidc_provider_hostpath = replace(local.oidc_provider_url, "https://", "") oidc_provider_arn = "arnawsiam::${data.aws_caller_identity.current.account_id}:oidc-provider/${local.oidc_provider_hostpath}" } # ---- S3 policy (scoped to bucket/prefix) ---- data "aws_iam_policy_document" "s3_rw" { statement { effect = "Allow" actions = ["s3:ListBucket"] resources = ["arnawss3:::${var.project_bucket}"] condition { test = "StringLike" variable = "s3:prefix" values = [var.project_prefix] } } statement { effect = "Allow" actions = ["s3:GetObject","s3:PutObject","s3:DeleteObject"] resources = ["arnawss3:::${var.project_bucket}/${var.project_prefix}"] } } # ---- Athena policy (workgroup + results bucket) ---- data "aws_iam_policy_document" "athena" { statement { effect = "Allow" actions = ["athena:StartQueryExecution","athena:GetQueryExecution","athena:GetQueryResults","athena:StopQueryExecution"] resources = [var.athena_workgroup_arn] } statement { effect = "Allow" actions = ["s3:PutObject","s3:GetObject","s3:ListBucket"] resources = [ "arnawss3:::${var.athena_results_bucket}", "arnawss3:::${var.athena_results_bucket}/${var.athena_results_prefix}" ] } } # ---- EMR Serverless policy (scope to application) ---- data "aws_iam_policy_document" "emr_serverless" { statement { effect = "Allow" actions = [ "emr-serverless:StartJobRun", "emr-serverless:GetJobRun", "emr-serverless:ListJobRuns", "emr-serverless:CancelJobRun" ] resources = [var.emr_serverless_app_arn] } } # ---- ALTERNATIVE: EMR on EC2 (classic) minimal example ---- # Uncomment and attach this instead of emr_serverless if you use EMR on EC2 # data "aws_iam_policy_document" "emr_ec2" { # statement { # effect = "Allow" # actions = [ # "elasticmapreduce:RunJobFlow", # "elasticmapreduce:AddJobFlowSteps", # "elasticmapreduce:Describe*", # "elasticmapreduce:List*", # "elasticmapreduce:TerminateJobFlows" # ] # resources = ["*"] # } # # If your flow creates clusters that assume a service role, you may need: # # statement { # # effect = "Allow" # # actions = ["iam:PassRole"]

# # resources = ["arnawsiam:ACCOUNT IDrole/EMR_DefaultRole"] # Adjust # # } # } # ---- Trust policy for IRSA (per ServiceAccount) ---- # Flow1 role (S3 + EMR) data "aws_iam_policy_document" "assume_role_flow1" { statement { effect = "Allow" actions = ["sts:AssumeRoleWithWebIdentity"] principals { type = "Federated" identifiers = [local.oidc_provider_arn] } condition { test = "StringEquals" variable = "${local.oidc_provider_hostpath}:aud" values = ["sts.amazonaws.com"] } condition { test = "StringEquals" variable = "${local.oidc_provider_hostpath}:sub" values = ["systemserviceaccount${var.namespace}:${var.sa_flow1}"] } } } resource "aws_iam_role" "flow1" { name = "prefect-flow1-role" assume_role_policy = data.aws_iam_policy_document.assume_role_flow1.json } # Inline policies for flow1 (S3 + EMR Serverless by default) resource "aws_iam_role_policy" "flow1_s3" { name = "s3-access" role = aws_iam_role.flow1.id policy = data.aws_iam_policy_document.s3_rw.json } resource "aws_iam_role_policy" "flow1_emr" { name = "emr-access" role = aws_iam_role.flow1.id policy = data.aws_iam_policy_document.emr_serverless.json # If using EMR on EC2 instead, use: data.aws_iam_policy_document.emr_ec2.json } # Flow2 role (S3 + Athena) data "aws_iam_policy_document" "assume_role_flow2" { statement { effect = "Allow" actions = ["sts:AssumeRoleWithWebIdentity"] principals { type = "Federated" identifiers = [local.oidc_provider_arn] } condition { test = "StringEquals" variable = "${local.oidc_provider_hostpath}:aud" values = ["sts.amazonaws.com"] } condition { test = "StringEquals" variable = "${local.oidc_provider_hostpath}:sub" values = ["systemserviceaccount${var.namespace}:${var.sa_flow2}"] } } } resource "aws_iam_role" "flow2" { name = "prefect-flow2-role" assume_role_policy = data.aws_iam_policy_document.assume_role_flow2.json } resource "aws_iam_role_policy" "flow2_s3" { name = "s3-access" role = aws_iam_role.flow2.id policy = data.aws_iam_policy_document.s3_rw.json } resource "aws_iam_role_policy" "flow2_athena" { name = "athena-access" role = aws_iam_role.flow2.id policy = data.aws_iam_policy_document.athena.json } output "flow1_role_arn" { value = aws_iam_role.flow1.arn } output "flow2_role_arn" { value = aws_iam_role.flow2.arn }

2) Kubernetes ServiceAccounts (annotated for IRSA)
Apply in the same namespace your Prefect runs in. Replace role ARNs with the outputs from Terraform.

apiVersion: v1 kind: ServiceAccount metadata: name: prefect-flow1-sa namespace: your-ns annotations: eks.amazonaws.com/role-arn: arnawsiam:123456789012role/prefect-flow1-role --- apiVersion: v1 kind: ServiceAccount metadata: name: prefect-flow2-sa namespace: your-ns annotations: eks.amazonaws.com/role-arn: arnawsiam:123456789012role/prefect-flow2-role

3) Prefect deployments: point each flow to its ServiceAccount
Option A: prefect.yaml

deployments: - name: flow1-s3-emr entrypoint: "flows/flow1.py:flow" work_pool: name: "kubernetes" job_variables: namespace: "your-ns" service_account_name: "prefect-flow1-sa" - name: flow2-s3-athena entrypoint: "flows/flow2.py:flow" work_pool: name: "kubernetes" job_variables: namespace: "your-ns" service_account_name: "prefect-flow2-sa"

Option B: Python deploy

from prefect import flow # flow1 flow.from_source( source="https://github.com/your-org/your-repo", entrypoint="flows/flow1.py:flow", ).deploy( name="flow1-s3-emr", work_pool_name="kubernetes", job_variables={"namespace": "your-ns", "service_account_name": "prefect-flow1-sa"}, ) # flow2 flow.from_source( source="https://github.com/your-org/your-repo",

entrypoint="flows/flow2.py:flow", ).deploy( name="flow2-s3-athena", work_pool_name="kubernetes", job_variables={"namespace": "your-ns", "service_account_name": "prefect-flow2-sa"}, ) ``` Operational tips - Keep the Prefect Kubernetes worker’s own ServiceAccount minimal — it just needs to create/watch Jobs, not access S3/EMR/Athena. - Lock each IAM role to a single ServiceAccount using the trust policy

sub

condition (as shown) to prevent privilege escalation. - Use Prefect Cloud RBAC to restrict who can edit deployments/work pools so users can’t swap

service_account_name

. Docs and references - Prefect: Kubernetes deployment guide, Customize job variables, Customize base job templates, Work pools concepts, Kubernetes worker integration - AWS: EKS IRSA (IAM roles for service accounts) Want me to tailor the IAM actions further? Let me know: - EMR flavor (Serverless, on EC2, or on EKS) - Exact S3 bucket/prefixes - Athena workgroup and results bucket/prefix

Got it — here’s an AWS-native (no Terraform) setup for IRSA with EKS, plus how to point Prefect to the service account. Answer to your IRSA question: - IRSA is the mechanism (OIDC provider + IAM role trust policy). You create the OIDC provider once per EKS cluster. - Each Kubernetes service account can assume exactly one IAM role (via the

<http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>

annotation). - Create as many IAM roles as you need for different permission sets and map them to different service accounts. You can reuse a role across multiple SAs if they need the same permissions. AWS CLI + manifests (replace placeholders in ALL_CAPS):

# Prereqs
export ACCOUNT_ID=<123456789012>
export REGION=<us-east-1>
export CLUSTER=<your-eks-cluster-name>
export NAMESPACE=<prefect>
export SA_NAME=<prefect-runner>
export ROLE_NAME=<PrefectRunnerRole>
export POLICY_NAME=<PrefectRunnerPolicy>

# 1) Get the cluster OIDC issuer (needed for IRSA)
OIDC_URL=$(aws eks describe-cluster --name "$CLUSTER" --region "$REGION" --query "cluster.identity.oidc.issuer" --output text)
OIDC_HOST=${OIDC_URL#https://}
echo "OIDC issuer: $OIDC_URL"

# 2) Check if an IAM OIDC provider already exists for this issuer
PROVIDER_ARN=$(aws iam list-open-id-connect-providers --query "OpenIDConnectProviderList[].Arn" --output text | tr '\t' '\n' | while read arn; do
  url=$(aws iam get-open-id-connect-provider --open-id-connect-provider-arn "$arn" --query 'Url' --output text 2>/dev/null || true)
  if [ "$url" = "$OIDC_HOST" ]; then echo "$arn"; fi
done)

if [ -z "$PROVIDER_ARN" ]; then
  echo "No OIDC provider found for $OIDC_HOST."
  echo "Create one per AWS docs (you'll need the root CA thumbprint):"
  echo "aws iam create-open-id-connect-provider --url $OIDC_URL --client-id-list <http://sts.amazonaws.com|sts.amazonaws.com> --thumbprint-list <THUMBPRINT>"
  echo "Docs: <https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html>"
  exit 1
else
  echo "Found OIDC provider: $PROVIDER_ARN"
fi

# 3) Create an IAM policy with the permissions your flows need (example: S3 read/write to one bucket)
cat > policy.json <<'EOF'
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3Access",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::<YOUR_BUCKET>",
        "arn:aws:s3:::<YOUR_BUCKET>/*"
      ]
    }
  ]
}
EOF

aws iam create-policy --policy-name "$POLICY_NAME" --policy-document <file://policy.json>

# 4) Create an IAM role trusted by the EKS OIDC provider, restricted to your service account and namespace
cat > trust.json <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "Federated": "$PROVIDER_ARN" },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "StringEquals": {
          "$OIDC_HOST:sub": "system:serviceaccount:$NAMESPACE:$SA_NAME",
          "$OIDC_HOST:aud": "sts.amazonaws.com"
        }
      }
    }
  ]
}
EOF

aws iam create-role --role-name "$ROLE_NAME" --assume-role-policy-document file://trust.json

# 5) Attach the permissions policy to the role
aws iam attach-role-policy --role-name "$ROLE_NAME" --policy-arn "arn:aws:iam::$ACCOUNT_ID:policy/$POLICY_NAME"

# 6) Create the Kubernetes namespace (if it doesn't exist)
kubectl get namespace "$NAMESPACE" >/dev/null 2>&1 || kubectl create namespace "$NAMESPACE"

# 7) Create the Kubernetes service account annotated with the IAM role ARN
cat <<EOF | kubectl apply -f -
apiVersion: v1
kind: ServiceAccount
metadata:
  name: $SA_NAME
  namespace: $NAMESPACE
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam::$ACCOUNT_ID:role/$ROLE_NAME
EOF

Tell Prefect to use this service account (Prefect 3.x): - In your deployment or work pool, set

job_variables.service_account_name

to the SA and optionally `namespace`: ``` # prefect.yaml deployments: - name: my-deployment entrypoint: flows/my_flow.py:my_flow work_pool:

name: kubernetes job_variables: namespace: "<prefect>" service_account_name: "<prefect-runner>"

Or in Python:

my_flow.from_source( source="https://github.com/your-org/your-repo", entrypoint="flows/my_flow.py:my_flow" ).deploy( name="my-deployment", work_pool_name="kubernetes", job_variables={"namespace": "<prefect>", "service_account_name": "<prefect-runner>"} ) ``` Notes: - One EKS OIDC provider per cluster; many IAM roles as needed; one role per service account (but the same role can be reused across SAs). - Keep roles least-privileged per workload/deployment by creating multiple service accounts + roles if your flows have different AWS needs. Links: - Prefect 3.x: How to run flows on Kubernetes and Override job variables (use

service_account_name

) - AWS IRSA: IAM roles for service accounts (IRSA) If you prefer this in CloudFormation or CDK instead of CLI/manifests, tell me your format and I’ll translate it.

Short answer: only if you need different AWS permission sets. - One EKS OIDC provider per cluster. - Each Kubernetes service account can reference exactly one IAM role (via the

<http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>

annotation). - So: - If all your Prefect flows need the same AWS permissions, use a single service account + single IAM role and reuse it across deployments. - If different flows need different permissions (least privilege), create multiple IAM roles and multiple service accounts, one SA per role, and point each deployment at the right SA. Example: two permission sets, two SAs and roles Kubernetes service accounts:

apiVersion: v1
kind: ServiceAccount
metadata:
  name: s3-readonly-sa
  namespace: prefect
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam::<ACCOUNT_ID>:role/S3ReadOnlyRole
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: s3-rw-sa
  namespace: prefect
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: arn:aws:iam::<ACCOUNT_ID>:role/S3ReadWriteRole

Prefect 3.x deployments pointing to different SAs:

deployments:
- name: flow-readonly
  entrypoint: flows/readonly.py:flow
  work_pool:
    name: kubernetes
    job_variables:
      namespace: "prefect"
      service_account_name: "s3-readonly-sa"

- name: flow-readwrite
  entrypoint: flows/readwrite.py:flow
  work_pool:
    name: kubernetes
    job_variables:
      namespace: "prefect"
      service_account_name: "s3-rw-sa"

Notes: - You can reuse the same IAM role across multiple SAs if they need the same permissions. - You cannot switch roles per pod without changing the SA annotation; use distinct SAs per permission set. - Docs: EKS IRSA, Prefect on Kubernetes, Prefect job variables (use

service_account_name

).

Great question. The Snowflake equivalent of the “one SA per permission set” pattern is “one Snowflake user + role per permission set,” and deliver those credentials to your Prefect pods via Kubernetes Secrets (or AWS Secrets Manager + External Secrets + IRSA). Then, point each Prefect deployment at the right secret. Recommended patterns (pick one): - Simple: Kubernetes Secret per workload, with a dedicated Snowflake user/role. Prefect loads it via env or envFrom. - Production: Store creds in AWS Secrets Manager; sync to K8s via External Secrets Operator (ESO) using IRSA; Prefect loads from the synced Secret. Keep least-privilege by creating multiple Snowflake users/roles and separate secrets. Auth options for Snowflake (machine-to-machine): - Key pair auth (recommended): per-workload Snowflake user with RSA key pair; store private key securely. - OAuth client credentials: use your IdP to get a token; store client_id/secret securely. - Username/password: possible, but less ideal. Quick-start: key pair auth with a K8s Secret 1) Create Snowflake role and user

-- as ACCOUNTADMIN (or role with appropriate privileges)
CREATE ROLE etl_readonly;
GRANT USAGE ON WAREHOUSE COMPUTE_WH TO ROLE etl_readonly;
GRANT USAGE ON DATABASE MYDB TO ROLE etl_readonly;
GRANT USAGE ON SCHEMA MYDB.PUBLIC TO ROLE etl_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA MYDB.PUBLIC TO ROLE etl_readonly;
GRANT ROLE etl_readonly TO USER etl_ro_user;

-- Create a user bound to that role
CREATE USER etl_ro_user
  DEFAULT_ROLE = etl_readonly
  DEFAULT_WAREHOUSE = COMPUTE_WH
  MUST_CHANGE_PASSWORD = FALSE;

2) Generate an RSA key pair and set the public key on the user

# Generate a 2048-bit RSA key pair and convert to PKCS8 (required by Snowflake Python)
openssl genrsa -out sf_key.pem 2048
openssl pkcs8 -topk8 -inform PEM -outform DER -in sf_key.pem -out sf_key_pk8.der -nocrypt
# (Optional) encrypt the private key instead, then store passphrase too

# Convert public key and register with Snowflake
openssl rsa -in sf_key.pem -pubout -out sf_key.pub
-- In Snowflake:
ALTER USER etl_ro_user SET RSA_PUBLIC_KEY='<contents of sf_key.pub without headers/footers>';

3) Store creds in a Kubernetes Secret (simple path)

apiVersion: v1
kind: Secret
metadata:
  name: snowflake-etl-ro
  namespace: prefect
type: Opaque
stringData:
  SNOWFLAKE_ACCOUNT: "<your_account_identifier>"     # e.g., xy12345.us-east-1
  SNOWFLAKE_USER: "etl_ro_user"
  SNOWFLAKE_ROLE: "etl_readonly"
  SNOWFLAKE_WAREHOUSE: "COMPUTE_WH"
  SNOWFLAKE_DATABASE: "MYDB"
  SNOWFLAKE_SCHEMA: "PUBLIC"
  # Base64-encode the PKCS8 DER bytes if you prefer, or mount as file.
  SNOWFLAKE_PRIVATE_KEY_B64: "<base64_of_sf_key_pk8.der>"

4) Tell Prefect to load this Secret into the job’s env - In Prefect 3.x, add envFrom via your Kubernetes work pool’s base job template (once), then pick which Secret per deployment via a variable. Base job template snippet (in the work pool) with a variable:

{
  "variables": {
    "snowflake_secret_name": {
      "title": "Snowflake K8s Secret Name",
      "type": "string",
      "default": "snowflake-etl-ro"
    }
  },
  "job_configuration": {
    "job_manifest": {
      "spec": {
        "template": {
          "spec": {
            "containers": [
              {
                "name": "prefect-job",
                "envFrom": [
                  {
                    "secretRef": {
                      "name": "{{ snowflake_secret_name }}"
                    }
                  }
                ]
              }
            ]
          }
        }
      }
    }
  }
}

Then pick the secret per deployment:

deployments:
- name: my-flow-ro
  entrypoint: flows/my_flow.py:flow
  work_pool:
    name: kubernetes
    job_variables:
      namespace: "prefect"
      service_account_name: "prefect-runner"   # if you also need IRSA for AWS calls
      snowflake_secret_name: "snowflake-etl-ro"

5) Use the Snowflake connector in your flow ``` import base64 import snowflake.connector def _load_private_key_from_env():

pk_der_b64 = os.environ["SNOWFLAKE_PRIVATE_KEY_B64"] return base64.b64decode(pk_der_b64) def get_snowflake_conn(): private_key = _load_private_key_from_env() return snowflake.connector.connect( account=os.environ["SNOWFLAKE_ACCOUNT"], user=os.environ["SNOWFLAKE_USER"], role=os.environ.get("SNOWFLAKE_ROLE"), warehouse=os.environ.get("SNOWFLAKE_WAREHOUSE"), database=os.environ.get("SNOWFLAKE_DATABASE"), schema=os.environ.get("SNOWFLAKE_SCHEMA"), private_key=private_key, # PKCS8 DER bytes authenticator="snowflake", # default; explicit for clarity ) ``` Production alternative: AWS Secrets Manager + External Secrets + IRSA - Store the same values in AWS Secrets Manager. - Use External Secrets Operator to sync to a K8s Secret your pod can read. - Give your workload’s Kubernetes service account an IAM role (IRSA) with “secretsmanager:GetSecretValue” for only the secrets it needs, mirroring the least-privilege pattern. - Point each Prefect deployment at a different

snowflake_secret_name

. OAuth variant (machine-to-machine) - Set up a Snowflake security integration with your IdP for client credentials. - Store `client_id`/`client_secret` in Secrets Manager. - At runtime, fetch a token, then connect with

authenticator='oauth', token=<access_token>

. Keep scopes/allowed roles constrained per workload. Docs: Snowflake OAuth overview, External OAuth. Do I need multiple service accounts? - For Snowflake itself: create multiple Snowflake users/roles if you want different permission sets. - For Kubernetes/AWS: only create multiple K8s service accounts if you are also pulling secrets from AWS (ESO/Secrets Manager) and want to lock each deployment to just its own secret ARNs via IRSA. If you’re using plain K8s Secrets, one SA is fine; just reference different Secret names per deployment. Helpful links: - Prefect: Run flows on Kubernetes and Customize base job templates (use envFrom as shown) - Snowflake: Key pair auth, Python connector, RBAC/roles - ESO: External Secrets Operator - AWS Secrets Manager: Secrets Manager intro If you tell me which auth you prefer (key pair vs OAuth) and whether you want AWS Secrets Manager in the loop, I can tailor the exact manifests/CLI for your setup.