Amazon Managed Workflows for Apache Airflow (Amazon MWAA) offers sturdy orchestration capabilities for information workflows, however managing DAG permissions at scale presents important operational challenges. As organizations develop their workflow environments and groups, manually assigning and sustaining person permissions turns into a bottleneck that may affect each safety and productiveness.

Conventional approaches require directors to manually configure role-based entry management (RBAC) for every DAG, resulting in:

• Inconsistent permission assignments throughout groups
• Delayed entry provisioning for brand spanking new crew members
• Elevated danger of human error in permission administration
• Vital operational overhead that doesn’t scale

There may be one other means of doing it by defining customized RBAC roles as talked about on this Amazon MWAA Consumer Information. Nonetheless, it doesn’t use Airflow tags to take action.

On this submit, we present you find out how to use Apache Airflow tags to systematically handle DAG permissions, lowering operational burden whereas sustaining sturdy safety controls that complement infrastructure-level safety measures.

Conditions

To implement this answer, you want:

AWS assets:

• An Amazon MWAA surroundings (model 2.7.2 or later, not supported in Airflow 3.0)
• IAM roles configured for Amazon MWAA entry with applicable belief relationships
• Amazon Easy Storage Service (Amazon S3) bucket for Amazon MWAA DAG storage with correct permissions

Permissions:

• IAM permissions to create and modify Amazon MWAA net login tokens
• Amazon MWAA execution position with permissions to entry the Apache Airflow metadata database
• Administrative entry to configure Apache Airflow roles and permissions

Answer overview

The automated permission administration system consists of 4 key elements that work collectively to offer scalable, safe entry management.The next diagram exhibits the workflow of how the answer works.

1. IAM integration layer – AWS IAM roles map on to Apache Airflow roles. Then, customers authenticate by means of AWS IAM and are mechanically assigned corresponding Airflow roles. This helps each particular person person roles and group-based entry patterns.
   
   Notice:
   • IAM Based mostly entry management to Amazon MWAA works for Apache Airflow default roles. For customized roles, the Admin person can assign the customized position utilizing the Apache Airflow UI as talked about within the Data Middle submit and within the Amazon MWAA Consumer Information.
   • If utilizing different authenticators, the tag-based DAG permissions proceed to work as said within the AWS Large Information Weblog submit.
2. Tag-based configuration – Apache Airflow tags outlined in DAGs are used to declare entry necessities. It helps read-only, edit, and delete permissions.
3. Automated synchronization engine – Scheduled DAG scans all energetic DAGs for permission tags primarily based on CRON schedule. It then processes tags and updates Apache Airflow RBAC permissions accordingly. Then, it offers a configuration primarily based to regulate the clean-up of current permissions.
4. Position-based entry management enforcement – Apache Airflow RBAC enforces the configured permissions by storing on Apache Airflow position and permissions metadata tables. Customers see solely the DAGs that they’ve entry to. They’ve granular management over learn in comparison with edit permissions.

Information circulation

1. Amazon MWAA Consumer assumes an IAM position to entry the Amazon MWAA UI.
2. DAG developer provides related tags to the DAG definition.
3. manage_dag_permissions DAG deployed to the Amazon MWAA surroundings runs on a CRON schedule, for instance, every day.
4. The DAG updates the respective position permissions to the DAG by updating the Apache Airflow metadata on the Apache Airflow DB.
5. Customers achieve or lose entry primarily based on their assigned roles.

Our answer builds upon the prevailing IAM integration of Amazon MWAA, whereas extending performance by means of customized automation:

1. Authentication and position mapping – Customers authenticate by means of AWS IAM roles that map on to corresponding Airflow roles.
2. Automated person creation – Upon first login, customers are mechanically created within the Apache Airflow metadata database with applicable position assignments.
3. Tag-based permission management – Every Apache Airflow position comprises particular DAG permissions primarily based on tags outlined within the DAGs.
4. Automated synchronization – A scheduled script maintains permissions as DAGs are added or modified.

Step 1: Configure IAM to Airflow position mapping

First, set up the mapping between your IAM principals and Apache Airflow roles. To grant permission utilizing the AWS Administration Console, full the next steps:

1. Check in to your AWS account and open the IAM console.
2. Within the left navigation pane, select Customers, then select your Amazon MWAA IAM person from the customers desk.
3. On the person particulars web page, beneath Abstract, select the Permissions tab, then select Permissions insurance policies to develop the cardboard and select Add permissions.
4. Within the Grant permissions part, select Connect current insurance policies straight, then select Create coverage to create and connect your personal customized permissions coverage.
5. On the Create coverage web page, select JSON, then copy and paste the next JSON permissions coverage within the coverage editor. This coverage grants net server entry to the person with the default Public Apache Airflow position.

{
  "Model": "2012-10-17",
  "Assertion": [
    {
      "Effect": "Allow",
      "Action": "airflow:CreateWebLoginToken",
      "Resource": "arn:aws:airflow:region:account-id:environment/your-environment-name"
    }
  ]
}

Step 2: Create the automated permission administration DAG

Now, create a DAG that may mechanically handle permissions primarily based on tags.

from airflow import DAG, settings
from airflow.operators.python import PythonOperator
from sqlalchemy import textual content
import pendulum
import logging

dag_id = "manage_dag_permissions"

class Constants:
    """
    Constants class to carry fixed values used all through the code.
    """
    AB_VIEW_MENU = "ab_view_menu"
    AB_PERMISSION = "ab_permission"
    AB_ROLE = "ab_role"
    AB_PERMISSION_VIEW = "ab_permission_view"
    AB_PERMISSION_VIEW_ROLE = "ab_permission_view_role"
    DAG_TAG = "dag_tag"

    CAN_READ = "can_read"
    CAN_EDIT = "can_edit"
    CAN_DELETE = "can_delete"


def _execute_query(sql_text, params=None, fetch=True):
    """
    Execute a parameterized SQL question towards the Airflow metadata DB.
    All queries use SQLAlchemy textual content() with bind parameters to stop SQL injection.

    Parameters:
        sql_text: SQL string with :named bind parameters
        params: dict of parameter values
        fetch: If True, return listing of first-column values; if False, commit and return None
    Returns:
        Listing of values (first column) if fetch=True, else None
    Raises:
        Re-raises any exception after rollback and logging
    """
    session = settings.Session()
    attempt:
        stmt = textual content(sql_text)
        if fetch:
            consequence = session.execute(stmt, params or {}).fetchall()
            return [row[0] for row in consequence]
        else:
            session.execute(stmt, params or {})
            session.commit()
            return None
    besides Exception as e:
        session.rollback()
        logging.error(f"DB question error (fetch={fetch}): {sort(e).__name__}: {e}")
        increase
    lastly:
        session.shut()

def fetch_airflow_role_id(role_name):
    """
    Fetch position id of a given position identify utilizing parameterized question.
    """
    consequence = _execute_query(
        "SELECT id FROM ab_role WHERE identify = :role_name",
        {"role_name": role_name},
    )
    if not consequence:
        increase ValueError(f"Airflow position not discovered: {role_name}")
    logging.information("Fetched position ID efficiently")
    return consequence[0]

def fetch_airflow_permission_id(permission_name):
    """
    Fetch permission id of a given permission utilizing parameterized question.
    """
    consequence = _execute_query(
        "SELECT id FROM ab_permission WHERE identify = :perm_name",
        {"perm_name": permission_name},
    )
    if not consequence:
        increase ValueError(f"Airflow permission not discovered: {permission_name}")
    logging.information("Fetched permission ID efficiently")
    return consequence[0]

def fetch_airflow_menu_object_ids(dag_names):
    """
    Fetch view_menu IDs for a listing of DAG useful resource names.
    Makes use of parameterized IN-clause through particular person bind params.

    Parameters:
        dag_names: listing of DAG useful resource names (e.g. ['DAG:my_dag1', 'DAG:my_dag2'])
    Returns:
        listing of view_menu IDs
    """
    if not dag_names:
        return []
    # Construct parameterized IN clause: :p0, :p1, :p2, ...
    param_names = [f":p{i}" for i in range(len(dag_names))]
    params = {f"p{i}": identify for i, identify in enumerate(dag_names)}
    in_clause = ", ".be a part of(param_names)
    consequence = _execute_query(
        f"SELECT id FROM ab_view_menu WHERE identify IN ({in_clause})",
        params,
    )
    logging.information(f"Fetched {len(consequence)} view menu IDs")
    return consequence

def fetch_perms_obj_association_ids(perm_id, view_menu_ids):
    """
    Fetch permission_view IDs for a permission and listing of view_menu IDs.
    Makes use of parameterized question.
    """
    if not view_menu_ids:
        return []
    param_names = [f":vm{i}" for i in range(len(view_menu_ids))]
    params = {f"vm{i}": vm_id for i, vm_id in enumerate(view_menu_ids)}
    params["perm_id"] = perm_id
    in_clause = ", ".be a part of(param_names)
    consequence = _execute_query(
        f"SELECT id FROM ab_permission_view WHERE permission_id = :perm_id AND view_menu_id IN ({in_clause})",
        params,
    )
    logging.information(f"Fetched {len(consequence)} permission-view affiliation IDs")
    return consequence

def fetch_dag_ids_by_tag(tag_name):
    """
    Fetch DAG IDs with a given tag identify utilizing parameterized question.
    """
    consequence = _execute_query(
        "SELECT DISTINCT dag_id FROM dag_tag WHERE identify = :tag_name",
        {"tag_name": tag_name},
    )
    logging.information(f"Fetched {len(consequence)} DAG IDs for tag")
    return consequence

def associate_permission_to_object(perm_id, view_menu_ids):
    """
    Affiliate permission to view_menu objects (DAGs) utilizing parameterized INSERT.
    """
    session = settings.Session()
    attempt:
        for vm_id in view_menu_ids:
            session.execute(
                textual content(
                    "INSERT INTO ab_permission_view (permission_id, view_menu_id) "
                    "VALUES (:perm_id, :vm_id) "
                    "ON CONFLICT (permission_id, view_menu_id) DO NOTHING"
                ),
                {"perm_id": perm_id, "vm_id": vm_id},
            )
        session.commit()
        logging.information(f"Related permission to {len(view_menu_ids)} view menus")
    besides Exception as e:
        session.rollback()
        logging.error(f"Error associating permission to things: {sort(e).__name__}: {e}")
        increase
    lastly:
        session.shut()

def associate_permission_to_role(permission_view_ids, role_id):
    """
    Affiliate permission_view entries to a task utilizing parameterized INSERT.
    """
    session = settings.Session()
    attempt:
        for pv_id in permission_view_ids:
            session.execute(
                textual content(
                    "INSERT INTO ab_permission_view_role (permission_view_id, role_id) "
                    "VALUES (:pv_id, :role_id) "
                    "ON CONFLICT (permission_view_id, role_id) DO NOTHING"
                ),
                {"pv_id": pv_id, "role_id": role_id},
            )
        session.commit()
        logging.information(f"Related {len(permission_view_ids)} permissions to position")
    besides Exception as e:
        session.rollback()
        logging.error(f"Error associating permissions to position: {sort(e).__name__}: {e}")
        increase
    lastly:
        session.shut()

def validate_if_permission_granted(permission_view_ids, role_id):
    """
    Validate if given permissions are related to given position utilizing parameterized question.
    """
    if not permission_view_ids:
        return []
    param_names = [f":pv{i}" for i in range(len(permission_view_ids))]
    params = {f"pv{i}": pv_id for i, pv_id in enumerate(permission_view_ids)}
    params["role_id"] = role_id
    in_clause = ", ".be a part of(param_names)
    consequence = _execute_query(
        f"SELECT id FROM ab_permission_view_role "
        f"WHERE permission_view_id IN ({in_clause}) AND role_id = :role_id",
        params,
    )
    logging.information(f"Validated {len(consequence)} permission grants")
    return consequence

def clean_up_existing_dag_permissions_for_role(role_id):
    """
    Clear up current DAG permissions for a given position utilizing parameterized question.
    Notice: this creates a quick window the place the position has no DAG permissions.
    """
    _execute_query(
        "DELETE FROM ab_permission_view_role WHERE id IN ("
        "  SELECT pvr.id"
        "  FROM ab_permission_view_role pvr"
        "  INNER JOIN ab_permission_view pv ON pvr.permission_view_id = pv.id"
        "  INNER JOIN ab_view_menu vm ON pv.view_menu_id = vm.id"
        "  WHERE pvr.role_id = :role_id AND vm.identify LIKE :dag_prefix"
        ")",
        {"role_id": role_id, "dag_prefix": "DAG:%"},
        fetch=False,
    )
    logging.information("Cleaned up current DAG permissions for position")

def sync_permission(config_data):
    """
    Sync permissions primarily based on the config.

    Parameters:
        config_data: dict with keys:
            - airflow_role_name: identify of the customized Airflow position
            - managed_dags: listing of DAG IDs to grant full administration permissions on
              (can_read, can_edit, can_delete)
            - do_cleanup: if True, take away all current DAG:* permissions first
    """
    # Get the position ID for position identify
    role_id = fetch_airflow_role_id(config_data["airflow_role_name"])

    # Clear up current DAG degree permissions if requested
    if config_data.get("do_cleanup", False):
        clean_up_existing_dag_permissions_for_role(role_id)

    managed_dags = config_data.get("managed_dags", [])
    if not managed_dags:
        logging.information("No managed DAGs discovered, skipping permission sync")
        return

    # Decide which permissions to grant (default: can_read solely)
    permissions = config_data.get("permissions", [Constants.CAN_READ])

    # Construct DAG useful resource names (e.g. ["DAG:my_dag1", "DAG:my_dag2"])
    dag_resource_names = [f"DAG:{dag.strip()}" for dag in managed_dags]

    # Get IDs for DAG view_menu objects
    vm_ids = fetch_airflow_menu_object_ids(dag_resource_names)
    if not vm_ids:
        logging.information("No view_menu entries discovered for managed DAGs")
        return

    # Grant the configured permissions on every managed DAG
    all_perm_view_ids = []
    for perm_name in permissions:
        perm_id = fetch_airflow_permission_id(perm_name)
        associate_permission_to_object(perm_id, vm_ids)
        all_perm_view_ids += fetch_perms_obj_association_ids(perm_id, vm_ids)

    # Affiliate permission_view entries with the position and validate
    if all_perm_view_ids and role_id:
        associate_permission_to_role(all_perm_view_ids, role_id)
        validate_if_permission_granted(all_perm_view_ids, role_id)

def sync_permissions_with_tags(role_mappings):
    """
    For every position mapping, fetch DAG IDs by tag and sync permissions.
    """
    for role_map in role_mappings:
        username = listing(role_map.keys())[0]
        airflow_role = role_map[username]["airflow_role"]
        edit_tag_name = role_map[username]["airflow_edit_tag"]

        config_data = {
            "airflow_role_name": airflow_role,
            "managed_dags": fetch_dag_ids_by_tag(edit_tag_name),
            "permissions": role_map[username].get("permissions", [Constants.CAN_READ]),
            "do_cleanup": role_map[username].get("do_cleanup", True),
        }
        logging.information(f"Syncing permissions for airflow position")
        sync_permission(config_data)
        logging.information("Accomplished permission sync for position")

"""
    Add new roles and permissions right here.
    Format:
    {
       "": {
            "airflow_role": ,
            "airflow_edit_tag": ,
            "permissions": ,
            "do_cleanup": 
        }
    },

    IMPORTANT - ROLE SETUP:
    When creating a brand new customized position (e.g. "analytics_reporting", "marketing_analyst")
    within the Airflow UI (Safety > Listing Roles), you MUST copy the Viewer position's
    permissions into the brand new position. The Viewer permissions present base UI entry
    (browse DAGs, view logs, menu entry, and so forth.). --or-- Assign the viewer position as effectively.
    With out them, customers assigned to
    the customized position will be unable to log in to the Airflow UI.

    This DAG manages DAG-level permissions on DAG:xxx assets.
    Which permissions are granted is managed by the "permissions" listing
    in every config entry (choices: can_read, can_edit, can_delete).
    It does NOT handle base UI permissions — these have to be arrange manually
    when creating the position.

    Steps to create a brand new customized position:
    1. Go to Safety > Listing Roles > + (Add)
    2. Identify it to match the "airflow_role" worth within the config under
    3. Copy all permissions from the "Viewer" position into the brand new position
    4. Save — this DAG will then mechanically add DAG-specific permissions
       (as configured within the "permissions" listing) for every tagged DAG
"""
role_mappings = [
    {
        "analytics_reporting": {
            "airflow_role": "analytics_reporting",
            "airflow_edit_tag": "analytics_reporting_edit",
            "permissions": ["can_read", "can_edit", "can_delete"],
            "do_cleanup": True,
        }
    },
    {
        "marketing_analyst": {
            "airflow_role": "marketing_analyst",
            "airflow_edit_tag": "marketing_analyst_edit",
            "permissions": ["can_read", "can_edit", "can_delete"],
            "do_cleanup": True,
        },
    },
]

with DAG(
    dag_id=dag_id,
    schedule="*/15 * * * *",
    catchup=False,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
) as dag:
    sync_dag_permissions_task = PythonOperator(
        task_id="sync_dag_permissions",
        python_callable=sync_permissions_with_tags,
        op_kwargs={"role_mappings": role_mappings},
    )

Step 3: Tag your DAGs for entry management

Add applicable tags to your DAGs to specify which roles ought to have entry. Tags are used to outline which roles have entry to tagged DAGs.

# Instance DAG for analytics_reporting
with DAG(
    "analytics_reporting_dag",
    description="Day by day analytics reporting pipeline",
    schedule_interval="@every day",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    catchup=False,
    tags=["reporting", "analytics", "analytics_reporting_edit"]
) as dag:
    # DAG duties right here
    move
    
    
# Instance DAG for marketing_analyst
with DAG(
    "marketing_analyst_dag",
    description="Day by day advertising lead evaluation pipeline",
    schedule_interval="@every day",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    catchup=False,
    tags=["marketing", "analytics", "marketing_analyst_edit"]
) as dag:
    # DAG duties right here
    move

On this instance:

• The analytics_reporting customized position may have learn, edit, and delete entry to the DAG analytics_reporting_dag (and different DAGs tagged with analytics_reporting_edit)
• The marketing_analyst customized position may have learn, edit, and delete entry to the DAG marketing_analyst_dag (and different DAGs tagged with marketing_analyst_edit)

The precise permissions granted (can_read, can_edit, can_delete) are configurable per position within the role_mappings config contained in the permission administration DAG:

role_mappings = [
    {
        "analytics_reporting": {
            "airflow_role": "analytics_reporting",
            "airflow_edit_tag": "analytics_reporting_edit",
            "permissions": ["can_read", "can_edit", "can_delete"],
            "do_cleanup": True,
        }
    },
    {
        "marketing_analyst": {
            "airflow_role": "marketing_analyst",
            "airflow_edit_tag": "marketing_analyst_edit",
            "permissions": ["can_read", "can_edit", "can_delete"],
            "do_cleanup": True,
        },
    },
]

Notice: Earlier than this DAG can handle permissions for a customized position, the position have to be created manually within the Apache Airflow UI (Safety > Listing Roles) with the Viewer position’s permissions copied in. See Step 2 for particulars.

Step 4: Deploy and check

1. Add each the permission administration DAG and your tagged DAGs to your Amazon MWAA surroundings’s S3 bucket.
2. Anticipate Amazon MWAA to detect and course of the brand new DAGs.
3. Confirm that the permission administration DAG runs efficiently.
4. Check entry with completely different person roles to substantiate correct permission enforcement.
5. Customers may also combine this with their CI/CD processes.

Troubleshooting

On this part, we cowl some frequent points and find out how to troubleshoot them.

Permission sync failures

Symptom: Permission sync DAG fails with database errors.

Trigger: Inadequate permissions on MWAA execution position.

Answer: Be certain that the execution position has airflow:CreateWebLoginToken permission and database entry.

Tags not being processed

Symptom: DAG tags are current however permissions aren’t up to date.

Answer: Examine that DAG is energetic and parsed efficiently – Overview permission sync DAG logs for processing errors.

Customers can’t entry anticipated DAGs

Symptom: Customers with right IAM roles can’t see DAGs

Answer: Verify that IAM to Apache Airflow position mapping is right. Confirm that the permission sync DAG has run efficiently. Examine Amazon CloudWatch Logs for permission task errors.

Efficiency points

Symptom: Permission sync takes too lengthy or occasions out.

Answer: Cut back sync frequency for giant environments. Contemplate batching permission updates. Monitor DAG execution time and optimize accordingly.

Debugging steps

1. Examine Amazon MWAA surroundings well being and connectivity
2. Overview permission sync DAG execution logs
3. Confirm IAM position configurations and belief relationships
4. Check with a single DAG to isolate points
5. Monitor CloudWatch Logs for detailed error messages

Advantages and concerns

Automated permission administration affords you important operational benefits whereas enhancing your safety. You’ll profit from diminished administrative overhead as handbook permission assignments are eliminated, so you’ll be able to scale seamlessly with out further burden. Your safety improves by means of constant software of least-privilege rules and diminished human error. You’ll improve your developer expertise with computerized entry provisioning that shortens onboarding time, whereas your system helps environments with over 500 DAGs with out efficiency degradation.

While you implement these programs, you could adhere to key safety practices. It is best to apply the precept of least privilege, validate tags to just be sure you’re solely processing approved tags, and set up complete audit mechanisms together with CloudTrail logging. Your entry management measures ought to prohibit permission administration capabilities to directors whilst you make the most of applicable position separation for various person personas.

You will want to think about a number of technical limitations throughout your implementation. IAM-based entry management to Amazon MWAA works solely with Apache Airflow default roles, not customized ones, although your tag-based permissions perform with various authenticators. Permission adjustments propagate primarily based on DAG schedules, doubtlessly inflicting delays. It is best to set up approval processes to your manufacturing adjustments, preserve model management for permissions, and doc your rollback procedures to make sure your system’s resilience and safety.

Clear up

Clear up assets after your experimentation:

1. Delete the Amazon MWAA environments utilizing the console or AWS CLI.
2. Replace the IAM position coverage or delete the IAM position if not wanted.

Conclusion

On this submit, you realized find out how to automate DAG permission administration in Amazon MWAA utilizing Apache Airflow’s tagging system. You noticed find out how to implement tag-based entry management that scales effectively, reduces handbook errors, and maintains least-privilege safety rules throughout a whole lot of DAGs. You additionally explored the important thing safety practices and technical concerns that you simply want to bear in mind throughout implementation.

Check out this answer in your Amazon MWAA surroundings to streamline your permission administration. Begin by implementing the tagging system in a growth surroundings, then steadily roll it out to manufacturing as your crew turns into comfy with the strategy.

In regards to the authors