diff --git a/airflow-core/docs/img/airflow_erd.sha256 b/airflow-core/docs/img/airflow_erd.sha256 index c1eb479c417d5..9ae7b32e1dd3d 100644 --- a/airflow-core/docs/img/airflow_erd.sha256 +++ b/airflow-core/docs/img/airflow_erd.sha256 @@ -1 +1 @@ -203aa3570578ef6e24b0f6725545e3ab830b349a9ca43f8f238ee9588245adc0 \ No newline at end of file +de526a7ff575b0f9deb174bf9e779bceb505d27a867bd2e8a67598de35f37c2f \ No newline at end of file diff --git a/airflow-core/docs/img/airflow_erd.svg b/airflow-core/docs/img/airflow_erd.svg index c7ef0ac5dd867..d1a70e1794108 100644 --- a/airflow-core/docs/img/airflow_erd.svg +++ b/airflow-core/docs/img/airflow_erd.svg @@ -12,567 +12,583 @@ job - -job - -id - - [INTEGER] - NOT NULL - -dag_id - - [VARCHAR(250)] - -end_date - - [TIMESTAMP] - -executor_class - - [VARCHAR(500)] - -hostname - - [VARCHAR(500)] - -job_type - - [VARCHAR(30)] - -latest_heartbeat - - [TIMESTAMP] - -start_date - - [TIMESTAMP] - -state - - [VARCHAR(20)] - -unixname - - [VARCHAR(1000)] + +job + +id + + [INTEGER] + NOT NULL + +dag_id + + [VARCHAR(250)] + +end_date + + [TIMESTAMP] + +executor_class + + [VARCHAR(500)] + +hostname + + [VARCHAR(500)] + +job_type + + [VARCHAR(30)] + +latest_heartbeat + + [TIMESTAMP] + +start_date + + [TIMESTAMP] + +state + + [VARCHAR(20)] + +unixname + + [VARCHAR(1000)] partitioned_asset_key_log - -partitioned_asset_key_log - -id - - [INTEGER] - NOT NULL - -asset_event_id - - [INTEGER] - NOT NULL - -asset_id - - [INTEGER] - NOT NULL - -asset_partition_dag_run_id - - [INTEGER] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -source_partition_key - - [VARCHAR(250)] - NOT NULL - -target_dag_id - - [VARCHAR(250)] - NOT NULL - -target_partition_key - - [VARCHAR(250)] - NOT NULL + +partitioned_asset_key_log + +id + + [INTEGER] + NOT NULL + +asset_event_id + + [INTEGER] + NOT NULL + +asset_id + + [INTEGER] + NOT NULL + +asset_partition_dag_run_id + + [INTEGER] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +source_partition_key + + [VARCHAR(250)] + NOT NULL + +target_dag_id + + [VARCHAR(250)] + NOT NULL + +target_partition_key + + [VARCHAR(250)] + NOT NULL log - -log - -id - - [INTEGER] - NOT NULL - -dag_id - - [VARCHAR(250)] - -dttm - - [TIMESTAMP] - NOT NULL - -event - - [VARCHAR(60)] - NOT NULL - -extra - - [TEXT] - -logical_date - - [TIMESTAMP] - -map_index - - [INTEGER] - -owner - - [VARCHAR(500)] - -owner_display_name - - [VARCHAR(500)] - -run_id - - [VARCHAR(250)] - -task_id - - [VARCHAR(250)] - -try_number - - [INTEGER] + +log + +id + + [INTEGER] + NOT NULL + +dag_id + + [VARCHAR(250)] + +dttm + + [TIMESTAMP] + NOT NULL + +event + + [VARCHAR(60)] + NOT NULL + +extra + + [TEXT] + +logical_date + + [TIMESTAMP] + +map_index + + [INTEGER] + +owner + + [VARCHAR(500)] + +owner_display_name + + [VARCHAR(500)] + +run_id + + [VARCHAR(250)] + +task_id + + [VARCHAR(250)] + +try_number + + [INTEGER] dag_priority_parsing_request - -dag_priority_parsing_request - -id - - [VARCHAR(32)] - NOT NULL - -bundle_name - - [VARCHAR(250)] - NOT NULL - -relative_fileloc - - [VARCHAR(2000)] - NOT NULL + +dag_priority_parsing_request + +id + + [VARCHAR(32)] + NOT NULL + +bundle_name + + [VARCHAR(250)] + NOT NULL + +relative_fileloc + + [VARCHAR(2000)] + NOT NULL import_error - -import_error - -id - - [INTEGER] - NOT NULL - -bundle_name - - [VARCHAR(250)] - -filename - - [VARCHAR(1024)] - -stacktrace - - [TEXT] - -timestamp - - [TIMESTAMP] + +import_error + +id + + [INTEGER] + NOT NULL + +bundle_name + + [VARCHAR(250)] + +filename + + [VARCHAR(1024)] + +stacktrace + + [TEXT] + +timestamp + + [TIMESTAMP] + + + +revoked_token + +revoked_token + +jti + + [VARCHAR(32)] + NOT NULL + +exp + + [TIMESTAMP] + NOT NULL - + dag_bundle - -dag_bundle - -name - - [VARCHAR(250)] - NOT NULL - -active - - [BOOLEAN] - -last_refreshed - - [TIMESTAMP] - -signed_url_template - - [VARCHAR(200)] - -template_params - - [JSON] - -version - - [VARCHAR(200)] + +dag_bundle + +name + + [VARCHAR(250)] + NOT NULL + +active + + [BOOLEAN] + +last_refreshed + + [TIMESTAMP] + +signed_url_template + + [VARCHAR(200)] + +template_params + + [JSON] + +version + + [VARCHAR(200)] - + dag_bundle_team - -dag_bundle_team - -dag_bundle_name - - [VARCHAR(250)] - NOT NULL - -team_name - - [VARCHAR(50)] - NOT NULL + +dag_bundle_team + +dag_bundle_name + + [VARCHAR(250)] + NOT NULL + +team_name + + [VARCHAR(50)] + NOT NULL - + dag_bundle:name--dag_bundle_team:dag_bundle_name - -0..N -1 + +0..N +1 - + dag - -dag - -dag_id - - [VARCHAR(250)] - NOT NULL - -asset_expression - - [JSON] - -bundle_name - - [VARCHAR(250)] - NOT NULL - -bundle_version - - [VARCHAR(200)] - -dag_display_name - - [VARCHAR(2000)] - -deadline - - [JSON] - -description - - [TEXT] - -exceeds_max_non_backfill - - [BOOLEAN] - NOT NULL - -fail_fast - - [BOOLEAN] - NOT NULL - -fileloc - - [VARCHAR(2000)] - -has_import_errors - - [BOOLEAN] - NOT NULL - -has_task_concurrency_limits - - [BOOLEAN] - NOT NULL - -is_paused - - [BOOLEAN] - NOT NULL - -is_stale - - [BOOLEAN] - NOT NULL - -last_expired - - [TIMESTAMP] - -last_parse_duration - - [DOUBLE PRECISION] - -last_parsed_time - - [TIMESTAMP] - -max_active_runs - - [INTEGER] - -max_active_tasks - - [INTEGER] - NOT NULL - -max_consecutive_failed_dag_runs - - [INTEGER] - NOT NULL - -next_dagrun - - [TIMESTAMP] - -next_dagrun_create_after - - [TIMESTAMP] - -next_dagrun_data_interval_end - - [TIMESTAMP] - -next_dagrun_data_interval_start - - [TIMESTAMP] - -owners - - [VARCHAR(2000)] - -relative_fileloc - - [VARCHAR(2000)] - -timetable_description - - [VARCHAR(1000)] - -timetable_summary - - [TEXT] - -timetable_type - - [VARCHAR(255)] - NOT NULL + +dag + +dag_id + + [VARCHAR(250)] + NOT NULL + +asset_expression + + [JSON] + +bundle_name + + [VARCHAR(250)] + NOT NULL + +bundle_version + + [VARCHAR(200)] + +dag_display_name + + [VARCHAR(2000)] + +deadline + + [JSON] + +description + + [TEXT] + +exceeds_max_non_backfill + + [BOOLEAN] + NOT NULL + +fail_fast + + [BOOLEAN] + NOT NULL + +fileloc + + [VARCHAR(2000)] + +has_import_errors + + [BOOLEAN] + NOT NULL + +has_task_concurrency_limits + + [BOOLEAN] + NOT NULL + +is_paused + + [BOOLEAN] + NOT NULL + +is_stale + + [BOOLEAN] + NOT NULL + +last_expired + + [TIMESTAMP] + +last_parse_duration + + [DOUBLE PRECISION] + +last_parsed_time + + [TIMESTAMP] + +max_active_runs + + [INTEGER] + +max_active_tasks + + [INTEGER] + NOT NULL + +max_consecutive_failed_dag_runs + + [INTEGER] + NOT NULL + +next_dagrun + + [TIMESTAMP] + +next_dagrun_create_after + + [TIMESTAMP] + +next_dagrun_data_interval_end + + [TIMESTAMP] + +next_dagrun_data_interval_start + + [TIMESTAMP] + +owners + + [VARCHAR(2000)] + +relative_fileloc + + [VARCHAR(2000)] + +timetable_description + + [VARCHAR(1000)] + +timetable_summary + + [TEXT] + +timetable_type + + [VARCHAR(255)] + NOT NULL dag_bundle:name--dag:bundle_name - -0..N -1 + +0..N +1 - + team - -team - -name - - [VARCHAR(50)] - NOT NULL + +team + +name + + [VARCHAR(50)] + NOT NULL - + team:name--dag_bundle_team:team_name - -0..N -1 + +0..N +1 - + connection - -connection - -id - - [INTEGER] - NOT NULL - -conn_id - - [VARCHAR(250)] - NOT NULL - -conn_type - - [VARCHAR(500)] - NOT NULL - -description - - [TEXT] - -extra - - [TEXT] - -host - - [VARCHAR(500)] - -is_encrypted - - [BOOLEAN] - NOT NULL - -is_extra_encrypted - - [BOOLEAN] - NOT NULL - -login - - [TEXT] - -password - - [TEXT] - -port - - [INTEGER] - -schema - - [VARCHAR(500)] - -team_name - - [VARCHAR(50)] + +connection + +id + + [INTEGER] + NOT NULL + +conn_id + + [VARCHAR(250)] + NOT NULL + +conn_type + + [VARCHAR(500)] + NOT NULL + +description + + [TEXT] + +extra + + [TEXT] + +host + + [VARCHAR(500)] + +is_encrypted + + [BOOLEAN] + NOT NULL + +is_extra_encrypted + + [BOOLEAN] + NOT NULL + +login + + [TEXT] + +password + + [TEXT] + +port + + [INTEGER] + +schema + + [VARCHAR(500)] + +team_name + + [VARCHAR(50)] team:name--connection:team_name - -0..N -{0,1} + +0..N +{0,1} - + slot_pool - -slot_pool - -id - - [INTEGER] - NOT NULL - -description - - [TEXT] - -include_deferred - - [BOOLEAN] - NOT NULL - -pool - - [VARCHAR(256)] - NOT NULL - -slots - - [INTEGER] - NOT NULL - -team_name - - [VARCHAR(50)] + +slot_pool + +id + + [INTEGER] + NOT NULL + +description + + [TEXT] + +include_deferred + + [BOOLEAN] + NOT NULL + +pool + + [VARCHAR(256)] + NOT NULL + +slots + + [INTEGER] + NOT NULL + +team_name + + [VARCHAR(50)] team:name--slot_pool:team_name - -0..N -{0,1} + +0..N +{0,1} - + variable - -variable - -id - - [INTEGER] - NOT NULL - -description - - [TEXT] - -is_encrypted - - [BOOLEAN] - NOT NULL - -key - - [VARCHAR(250)] - NOT NULL - -team_name - - [VARCHAR(50)] - -val - - [TEXT] - NOT NULL + +variable + +id + + [INTEGER] + NOT NULL + +description + + [TEXT] + +is_encrypted + + [BOOLEAN] + NOT NULL + +key + + [VARCHAR(250)] + NOT NULL + +team_name + + [VARCHAR(50)] + +val + + [TEXT] + NOT NULL team:name--variable:team_name - -0..N -{0,1} + +0..N +{0,1} - + trigger trigger @@ -606,7 +622,7 @@ [INTEGER] - + callback callback @@ -661,7 +677,7 @@ {0,1} - + asset_watcher asset_watcher @@ -689,184 +705,184 @@ 1 - + task_instance - -task_instance - -id - - [UUID] - NOT NULL - -context_carrier - - [JSONB] - -custom_operator_name - - [VARCHAR(1000)] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -dag_version_id - - [UUID] - -duration - - [DOUBLE PRECISION] - -end_date - - [TIMESTAMP] - -executor - - [VARCHAR(1000)] - -executor_config - - [BYTEA] - NOT NULL - -external_executor_id - - [TEXT] - -hostname - - [VARCHAR(1000)] - NOT NULL - -last_heartbeat_at - - [TIMESTAMP] - -map_index - - [INTEGER] - NOT NULL - -max_tries - - [INTEGER] - NOT NULL - -next_kwargs - - [JSONB] - -next_method - - [VARCHAR(1000)] - -operator - - [VARCHAR(1000)] - -pid - - [INTEGER] - -pool - - [VARCHAR(256)] - NOT NULL - -pool_slots - - [INTEGER] - NOT NULL - -priority_weight - - [INTEGER] - NOT NULL - -queue - - [VARCHAR(256)] - NOT NULL - -queued_by_job_id - - [INTEGER] - -queued_dttm - - [TIMESTAMP] - -rendered_map_index - - [VARCHAR(250)] - -run_id - - [VARCHAR(250)] - NOT NULL - -scheduled_dttm - - [TIMESTAMP] - -span_status - - [VARCHAR(250)] - NOT NULL - -start_date - - [TIMESTAMP] - -state - - [VARCHAR(20)] - -task_display_name - - [VARCHAR(2000)] - -task_id - - [VARCHAR(250)] - NOT NULL - -trigger_id - - [INTEGER] - -trigger_timeout - - [TIMESTAMP] - -try_number - - [INTEGER] - NOT NULL - -unixname - - [VARCHAR(1000)] - NOT NULL - -updated_at - - [TIMESTAMP] + +task_instance + +id + + [UUID] + NOT NULL + +context_carrier + + [JSONB] + +custom_operator_name + + [VARCHAR(1000)] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +dag_version_id + + [UUID] + +duration + + [DOUBLE PRECISION] + +end_date + + [TIMESTAMP] + +executor + + [VARCHAR(1000)] + +executor_config + + [BYTEA] + NOT NULL + +external_executor_id + + [TEXT] + +hostname + + [VARCHAR(1000)] + NOT NULL + +last_heartbeat_at + + [TIMESTAMP] + +map_index + + [INTEGER] + NOT NULL + +max_tries + + [INTEGER] + NOT NULL + +next_kwargs + + [JSONB] + +next_method + + [VARCHAR(1000)] + +operator + + [VARCHAR(1000)] + +pid + + [INTEGER] + +pool + + [VARCHAR(256)] + NOT NULL + +pool_slots + + [INTEGER] + NOT NULL + +priority_weight + + [INTEGER] + NOT NULL + +queue + + [VARCHAR(256)] + NOT NULL + +queued_by_job_id + + [INTEGER] + +queued_dttm + + [TIMESTAMP] + +rendered_map_index + + [VARCHAR(250)] + +run_id + + [VARCHAR(250)] + NOT NULL + +scheduled_dttm + + [TIMESTAMP] + +span_status + + [VARCHAR(250)] + NOT NULL + +start_date + + [TIMESTAMP] + +state + + [VARCHAR(20)] + +task_display_name + + [VARCHAR(2000)] + +task_id + + [VARCHAR(250)] + NOT NULL + +trigger_id + + [INTEGER] + +trigger_timeout + + [TIMESTAMP] + +try_number + + [INTEGER] + NOT NULL + +unixname + + [VARCHAR(1000)] + NOT NULL + +updated_at + + [TIMESTAMP] - + trigger:id--task_instance:trigger_id - -0..N + +0..N {0,1} - + deadline deadline @@ -910,412 +926,412 @@ NOT NULL - + callback:id--deadline:callback_id 0..N 1 - + asset_alias - -asset_alias - -id - - [INTEGER] - NOT NULL - -group - - [VARCHAR(1500)] - NOT NULL - -name - - [VARCHAR(1500)] - NOT NULL + +asset_alias + +id + + [INTEGER] + NOT NULL + +group + + [VARCHAR(1500)] + NOT NULL + +name + + [VARCHAR(1500)] + NOT NULL - + asset_alias_asset - -asset_alias_asset - -alias_id - - [INTEGER] - NOT NULL - -asset_id - - [INTEGER] - NOT NULL + +asset_alias_asset + +alias_id + + [INTEGER] + NOT NULL + +asset_id + + [INTEGER] + NOT NULL asset_alias:id--asset_alias_asset:alias_id - -0..N -1 + +0..N +1 - + asset_alias_asset_event - -asset_alias_asset_event - -alias_id - - [INTEGER] - NOT NULL - -event_id - - [INTEGER] - NOT NULL + +asset_alias_asset_event + +alias_id + + [INTEGER] + NOT NULL + +event_id + + [INTEGER] + NOT NULL asset_alias:id--asset_alias_asset_event:alias_id - -0..N -1 + +0..N +1 - + dag_schedule_asset_alias_reference - -dag_schedule_asset_alias_reference - -alias_id - - [INTEGER] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -updated_at - - [TIMESTAMP] - NOT NULL + +dag_schedule_asset_alias_reference + +alias_id + + [INTEGER] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +updated_at + + [TIMESTAMP] + NOT NULL - + asset_alias:id--dag_schedule_asset_alias_reference:alias_id - -0..N -1 + +0..N +1 - + asset - -asset - -id - - [INTEGER] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -extra - - [JSON] - NOT NULL - -group - - [VARCHAR(1500)] - NOT NULL - -name - - [VARCHAR(1500)] - NOT NULL - -updated_at - - [TIMESTAMP] - NOT NULL - -uri - - [VARCHAR(1500)] - NOT NULL + +asset + +id + + [INTEGER] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +extra + + [JSON] + NOT NULL + +group + + [VARCHAR(1500)] + NOT NULL + +name + + [VARCHAR(1500)] + NOT NULL + +updated_at + + [TIMESTAMP] + NOT NULL + +uri + + [VARCHAR(1500)] + NOT NULL asset:id--asset_alias_asset:asset_id - -0..N -1 + +0..N +1 asset:id--asset_watcher:asset_id - + 0..N -1 +1 - + asset_active - -asset_active - -name - - [VARCHAR(1500)] - NOT NULL - -uri - - [VARCHAR(1500)] - NOT NULL + +asset_active + +name + + [VARCHAR(1500)] + NOT NULL + +uri + + [VARCHAR(1500)] + NOT NULL asset:name--asset_active:name - -1 -1 + +1 +1 asset:uri--asset_active:uri - -1 -1 + +1 +1 - + dag_schedule_asset_reference - -dag_schedule_asset_reference - -asset_id - - [INTEGER] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -updated_at - - [TIMESTAMP] - NOT NULL + +dag_schedule_asset_reference + +asset_id + + [INTEGER] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +updated_at + + [TIMESTAMP] + NOT NULL asset:id--dag_schedule_asset_reference:asset_id - -0..N -1 + +0..N +1 - + task_outlet_asset_reference - -task_outlet_asset_reference - -asset_id - - [INTEGER] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -task_id - - [VARCHAR(250)] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -updated_at - - [TIMESTAMP] - NOT NULL + +task_outlet_asset_reference + +asset_id + + [INTEGER] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +task_id + + [VARCHAR(250)] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +updated_at + + [TIMESTAMP] + NOT NULL - + asset:id--task_outlet_asset_reference:asset_id - -0..N -1 + +0..N +1 - + task_inlet_asset_reference - -task_inlet_asset_reference - -asset_id - - [INTEGER] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -task_id - - [VARCHAR(250)] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -updated_at - - [TIMESTAMP] - NOT NULL + +task_inlet_asset_reference + +asset_id + + [INTEGER] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +task_id + + [VARCHAR(250)] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +updated_at + + [TIMESTAMP] + NOT NULL - + asset:id--task_inlet_asset_reference:asset_id - -0..N -1 + +0..N +1 - + asset_dag_run_queue - -asset_dag_run_queue - -asset_id - - [INTEGER] - NOT NULL - -target_dag_id - - [VARCHAR(250)] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL + +asset_dag_run_queue + +asset_id + + [INTEGER] + NOT NULL + +target_dag_id + + [VARCHAR(250)] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL - + asset:id--asset_dag_run_queue:asset_id - -0..N -1 + +0..N +1 - + asset_event - -asset_event - -id - - [INTEGER] - NOT NULL - -asset_id - - [INTEGER] - NOT NULL - -extra - - [JSON] - NOT NULL - -partition_key - - [VARCHAR(250)] - -source_dag_id - - [VARCHAR(250)] - -source_map_index - - [INTEGER] - -source_run_id - - [VARCHAR(250)] - -source_task_id - - [VARCHAR(250)] - -timestamp - - [TIMESTAMP] - NOT NULL + +asset_event + +id + + [INTEGER] + NOT NULL + +asset_id + + [INTEGER] + NOT NULL + +extra + + [JSON] + NOT NULL + +partition_key + + [VARCHAR(250)] + +source_dag_id + + [VARCHAR(250)] + +source_map_index + + [INTEGER] + +source_run_id + + [VARCHAR(250)] + +source_task_id + + [VARCHAR(250)] + +timestamp + + [TIMESTAMP] + NOT NULL asset_event:id--asset_alias_asset_event:event_id - -0..N -1 + +0..N +1 - + dagrun_asset_event - -dagrun_asset_event - -dag_run_id - - [INTEGER] - NOT NULL - -event_id - - [INTEGER] - NOT NULL + +dagrun_asset_event + +dag_run_id + + [INTEGER] + NOT NULL + +event_id + + [INTEGER] + NOT NULL - + asset_event:id--dagrun_asset_event:event_id - -0..N -1 + +0..N +1 - + dag_schedule_asset_name_reference dag_schedule_asset_name_reference @@ -1338,12 +1354,12 @@ dag:dag_id--dag_schedule_asset_name_reference:dag_id - + 0..N -1 +1 - + dag_schedule_asset_uri_reference dag_schedule_asset_uri_reference @@ -1366,93 +1382,93 @@ dag:dag_id--dag_schedule_asset_uri_reference:dag_id - + 0..N -1 +1 - + dag:dag_id--dag_schedule_asset_alias_reference:dag_id - -0..N -1 + +0..N +1 dag:dag_id--dag_schedule_asset_reference:dag_id - -0..N -1 + +0..N +1 - + dag:dag_id--task_outlet_asset_reference:dag_id - -0..N -1 + +0..N +1 - + dag:dag_id--task_inlet_asset_reference:dag_id - -0..N -1 + +0..N +1 - + dag:dag_id--asset_dag_run_queue:target_dag_id - -0..N -1 + +0..N +1 - + dag_version - -dag_version - -id - - [UUID] - NOT NULL - -bundle_name - - [VARCHAR(250)] - -bundle_version - - [VARCHAR(250)] - -created_at - - [TIMESTAMP] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -last_updated - - [TIMESTAMP] - NOT NULL - -version_number - - [INTEGER] - NOT NULL + +dag_version + +id + + [UUID] + NOT NULL + +bundle_name + + [VARCHAR(250)] + +bundle_version + + [VARCHAR(250)] + +created_at + + [TIMESTAMP] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +last_updated + + [TIMESTAMP] + NOT NULL + +version_number + + [INTEGER] + NOT NULL dag:dag_id--dag_version:dag_id - -0..N -1 + +0..N +1 - + dag_tag dag_tag @@ -1470,12 +1486,12 @@ dag:dag_id--dag_tag:dag_id - + 0..N -1 +1 - + dag_owner_attributes dag_owner_attributes @@ -1498,12 +1514,12 @@ dag:dag_id--dag_owner_attributes:dag_id - + 0..N -1 +1 - + dag_warning dag_warning @@ -1531,12 +1547,12 @@ dag:dag_id--dag_warning:dag_id - + 0..N -1 +1 - + dag_favorite dag_favorite @@ -1554,143 +1570,143 @@ dag:dag_id--dag_favorite:dag_id - + 0..N -1 +1 - + dag_run - -dag_run - -id - - [INTEGER] - NOT NULL - -backfill_id - - [INTEGER] - -bundle_version - - [VARCHAR(250)] - -clear_number - - [INTEGER] - NOT NULL - -conf - - [JSONB] - -context_carrier - - [JSONB] - -created_dag_version_id - - [UUID] - -creating_job_id - - [INTEGER] - -dag_id - - [VARCHAR(250)] - NOT NULL - -data_interval_end - - [TIMESTAMP] - -data_interval_start - - [TIMESTAMP] - -end_date - - [TIMESTAMP] - -last_scheduling_decision - - [TIMESTAMP] - -log_template_id - - [INTEGER] - NOT NULL - -logical_date - - [TIMESTAMP] - -partition_key - - [VARCHAR(250)] - -queued_at - - [TIMESTAMP] - -run_after - - [TIMESTAMP] - NOT NULL - -run_id - - [VARCHAR(250)] - NOT NULL - -run_type - - [VARCHAR(50)] - NOT NULL - -scheduled_by_job_id - - [INTEGER] - -span_status - - [VARCHAR(250)] - NOT NULL - -start_date - - [TIMESTAMP] - -state - - [VARCHAR(50)] - NOT NULL - -triggered_by - - [VARCHAR(50)] - -triggering_user_name - - [VARCHAR(512)] - -updated_at - - [TIMESTAMP] - NOT NULL + +dag_run + +id + + [INTEGER] + NOT NULL + +backfill_id + + [INTEGER] + +bundle_version + + [VARCHAR(250)] + +clear_number + + [INTEGER] + NOT NULL + +conf + + [JSONB] + +context_carrier + + [JSONB] + +created_dag_version_id + + [UUID] + +creating_job_id + + [INTEGER] + +dag_id + + [VARCHAR(250)] + NOT NULL + +data_interval_end + + [TIMESTAMP] + +data_interval_start + + [TIMESTAMP] + +end_date + + [TIMESTAMP] + +last_scheduling_decision + + [TIMESTAMP] + +log_template_id + + [INTEGER] + NOT NULL + +logical_date + + [TIMESTAMP] + +partition_key + + [VARCHAR(250)] + +queued_at + + [TIMESTAMP] + +run_after + + [TIMESTAMP] + NOT NULL + +run_id + + [VARCHAR(250)] + NOT NULL + +run_type + + [VARCHAR(50)] + NOT NULL + +scheduled_by_job_id + + [INTEGER] + +span_status + + [VARCHAR(250)] + NOT NULL + +start_date + + [TIMESTAMP] + +state + + [VARCHAR(50)] + NOT NULL + +triggered_by + + [VARCHAR(50)] + +triggering_user_name + + [VARCHAR(512)] + +updated_at + + [TIMESTAMP] + NOT NULL - + dag_version:id--dag_run:created_dag_version_id - -0..N -{0,1} + +0..N +{0,1} - + dag_code dag_code @@ -1738,12 +1754,12 @@ dag_version:id--dag_code:dag_version_id - + 0..N -1 +1 - + serialized_dag serialized_dag @@ -1789,115 +1805,115 @@ dag_version:id--serialized_dag:dag_version_id - + 0..N -1 +1 - + dag_version:id--task_instance:dag_version_id - -0..N -{0,1} + +0..N +{0,1} - + log_template - -log_template - -id - - [INTEGER] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -elasticsearch_id - - [TEXT] - NOT NULL - -filename - - [TEXT] - NOT NULL + +log_template + +id + + [INTEGER] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +elasticsearch_id + + [TEXT] + NOT NULL + +filename + + [TEXT] + NOT NULL - + log_template:id--dag_run:log_template_id - -0..N -1 + +0..N +1 - + dag_run:id--dagrun_asset_event:dag_run_id - -0..N -1 + +0..N +1 - + asset_partition_dag_run - -asset_partition_dag_run - -id - - [INTEGER] - NOT NULL - -created_at - - [TIMESTAMP] - NOT NULL - -created_dag_run_id - - [INTEGER] - -partition_key - - [VARCHAR(250)] - NOT NULL - -target_dag_id - - [VARCHAR(250)] - NOT NULL - -updated_at - - [TIMESTAMP] - NOT NULL + +asset_partition_dag_run + +id + + [INTEGER] + NOT NULL + +created_at + + [TIMESTAMP] + NOT NULL + +created_dag_run_id + + [INTEGER] + +partition_key + + [VARCHAR(250)] + NOT NULL + +target_dag_id + + [VARCHAR(250)] + NOT NULL + +updated_at + + [TIMESTAMP] + NOT NULL dag_run:id--asset_partition_dag_run:created_dag_run_id - -0..N -{0,1} + +0..N +{0,1} - -dag_run:dag_id--task_instance:dag_id - -0..N -1 + +dag_run:run_id--task_instance:run_id + +0..N +1 - -dag_run:run_id--task_instance:run_id - -0..N -1 + +dag_run:dag_id--task_instance:dag_id + +0..N +1 - + backfill_dag_run backfill_dag_run @@ -1933,132 +1949,132 @@ dag_run:id--backfill_dag_run:dag_run_id - + 0..N -{0,1} +{0,1} - + dag_run_note - -dag_run_note - -dag_run_id - - [INTEGER] - NOT NULL - -content - - [VARCHAR(1000)] - -created_at - - [TIMESTAMP] - NOT NULL - -updated_at - - [TIMESTAMP] - NOT NULL - -user_id - - [VARCHAR(128)] + +dag_run_note + +dag_run_id + + [INTEGER] + NOT NULL + +content + + [VARCHAR(1000)] + +created_at + + [TIMESTAMP] + NOT NULL + +updated_at + + [TIMESTAMP] + NOT NULL + +user_id + + [VARCHAR(128)] dag_run:id--dag_run_note:dag_run_id - -1 -1 + +1 +1 - + dag_run:id--deadline:dagrun_id - + 0..N -{0,1} +{0,1} - + backfill - -backfill - -id - - [INTEGER] - NOT NULL - -completed_at - - [TIMESTAMP] - -created_at - - [TIMESTAMP] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -dag_run_conf - - [JSON] - NOT NULL - -from_date - - [TIMESTAMP] - NOT NULL - -is_paused - - [BOOLEAN] - -max_active_runs - - [INTEGER] - NOT NULL - -reprocess_behavior - - [VARCHAR(250)] - NOT NULL - -to_date - - [TIMESTAMP] - NOT NULL - -triggering_user_name - - [VARCHAR(512)] - -updated_at - - [TIMESTAMP] - NOT NULL + +backfill + +id + + [INTEGER] + NOT NULL + +completed_at + + [TIMESTAMP] + +created_at + + [TIMESTAMP] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +dag_run_conf + + [JSON] + NOT NULL + +from_date + + [TIMESTAMP] + NOT NULL + +is_paused + + [BOOLEAN] + +max_active_runs + + [INTEGER] + NOT NULL + +reprocess_behavior + + [VARCHAR(250)] + NOT NULL + +to_date + + [TIMESTAMP] + NOT NULL + +triggering_user_name + + [VARCHAR(512)] + +updated_at + + [TIMESTAMP] + NOT NULL - + backfill:id--dag_run:backfill_id - -0..N -{0,1} + +0..N +{0,1} backfill:id--backfill_dag_run:backfill_id - + 0..N -1 +1 - + deadline_alert deadline_alert @@ -2109,7 +2125,7 @@ 1 - + hitl_detail hitl_detail @@ -2175,75 +2191,75 @@ task_instance:id--hitl_detail:ti_id - + 1 -1 +1 - + task_map - -task_map - -dag_id - - [VARCHAR(250)] - NOT NULL - -map_index - - [INTEGER] - NOT NULL - -run_id - - [VARCHAR(250)] - NOT NULL - -task_id - - [VARCHAR(250)] - NOT NULL - -keys - - [JSONB] - -length - - [INTEGER] - NOT NULL + +task_map + +dag_id + + [VARCHAR(250)] + NOT NULL + +map_index + + [INTEGER] + NOT NULL + +run_id + + [VARCHAR(250)] + NOT NULL + +task_id + + [VARCHAR(250)] + NOT NULL + +keys + + [JSONB] + +length + + [INTEGER] + NOT NULL -task_instance:map_index--task_map:map_index - -0..N -1 +task_instance:dag_id--task_map:dag_id + +0..N +1 task_instance:task_id--task_map:task_id - -0..N -1 + +0..N +1 -task_instance:run_id--task_map:run_id - -0..N -1 +task_instance:map_index--task_map:map_index + +0..N +1 -task_instance:dag_id--task_map:dag_id - -0..N -1 +task_instance:run_id--task_map:run_id + +0..N +1 - + task_reschedule task_reschedule @@ -2281,85 +2297,85 @@ task_instance:id--task_reschedule:ti_id - + 0..N -1 +1 - + xcom - -xcom - -dag_run_id - - [INTEGER] - NOT NULL - -key - - [VARCHAR(512)] - NOT NULL - -map_index - - [INTEGER] - NOT NULL - -task_id - - [VARCHAR(250)] - NOT NULL - -dag_id - - [VARCHAR(250)] - NOT NULL - -run_id - - [VARCHAR(250)] - NOT NULL - -timestamp - - [TIMESTAMP] - NOT NULL - -value - - [JSONB] + +xcom + +dag_run_id + + [INTEGER] + NOT NULL + +key + + [VARCHAR(512)] + NOT NULL + +map_index + + [INTEGER] + NOT NULL + +task_id + + [VARCHAR(250)] + NOT NULL + +dag_id + + [VARCHAR(250)] + NOT NULL + +run_id + + [VARCHAR(250)] + NOT NULL + +timestamp + + [TIMESTAMP] + NOT NULL + +value + + [JSONB] -task_instance:task_id--xcom:task_id - -0..N -1 +task_instance:run_id--xcom:run_id + +0..N +1 -task_instance:run_id--xcom:run_id - -0..N -1 +task_instance:map_index--xcom:map_index + +0..N +1 -task_instance:map_index--xcom:map_index - -0..N -1 +task_instance:task_id--xcom:task_id + +0..N +1 task_instance:dag_id--xcom:dag_id - -0..N -1 + +0..N +1 - + task_instance_note task_instance_note @@ -2390,12 +2406,12 @@ task_instance:id--task_instance_note:ti_id - + 1 -1 +1 - + task_instance_history task_instance_history @@ -2556,103 +2572,103 @@ task_instance:map_index--task_instance_history:map_index - + 0..N -1 +1 -task_instance:dag_id--task_instance_history:dag_id - -0..N -1 +task_instance:run_id--task_instance_history:run_id + +0..N +1 -task_instance:run_id--task_instance_history:run_id - -0..N -1 +task_instance:dag_id--task_instance_history:dag_id + +0..N +1 task_instance:task_id--task_instance_history:task_id - + 0..N -1 +1 - + rendered_task_instance_fields - -rendered_task_instance_fields - -dag_id - - [VARCHAR(250)] - NOT NULL - -map_index - - [INTEGER] - NOT NULL + +rendered_task_instance_fields -run_id - - [VARCHAR(250)] - NOT NULL +dag_id + + [VARCHAR(250)] + NOT NULL -task_id - - [VARCHAR(250)] - NOT NULL +map_index + + [INTEGER] + NOT NULL -k8s_pod_yaml - - [JSON] +run_id + + [VARCHAR(250)] + NOT NULL -rendered_fields - - [JSON] - NOT NULL +task_id + + [VARCHAR(250)] + NOT NULL + +k8s_pod_yaml + + [JSON] + +rendered_fields + + [JSON] + NOT NULL -task_instance:run_id--rendered_task_instance_fields:run_id - -0..N -1 +task_instance:task_id--rendered_task_instance_fields:task_id + +0..N +1 task_instance:map_index--rendered_task_instance_fields:map_index - -0..N -1 + +0..N +1 -task_instance:dag_id--rendered_task_instance_fields:dag_id - -0..N -1 +task_instance:run_id--rendered_task_instance_fields:run_id + +0..N +1 -task_instance:task_id--rendered_task_instance_fields:task_id - -0..N -1 +task_instance:dag_id--rendered_task_instance_fields:dag_id + +0..N +1 - + deadline_alert:id--deadline:deadline_alert_id 0..N {0,1} - + hitl_detail_history hitl_detail_history @@ -2723,15 +2739,15 @@ 1 - + alembic_version - -alembic_version - -version_num - - [VARCHAR(32)] - NOT NULL + +alembic_version + +version_num + + [VARCHAR(32)] + NOT NULL diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst index 4f6892e6b3458..61cff3374e17b 100644 --- a/airflow-core/docs/migrations-ref.rst +++ b/airflow-core/docs/migrations-ref.rst @@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | Revision ID | Revises ID | Airflow Version | Description | +=========================+==================+===================+==============================================================+ -| ``a5a3e5eb9b8d`` (head) | ``82dbd68e6171`` | ``3.2.0`` | Make external_executor_id TEXT to allow for longer | +| ``53ff648b8a26`` (head) | ``a5a3e5eb9b8d`` | ``3.2.0`` | Add revoked_token table. | ++-------------------------+------------------+-------------------+--------------------------------------------------------------+ +| ``a5a3e5eb9b8d`` | ``82dbd68e6171`` | ``3.2.0`` | Make external_executor_id TEXT to allow for longer | | | | | external_executor_ids. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ | ``82dbd68e6171`` | ``55297ae24532`` | ``3.2.0`` | Add index to task_reschedule ti_id . | diff --git a/airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py b/airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py index 62cfbd72485de..122111a85e51d 100644 --- a/airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py +++ b/airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py @@ -45,6 +45,7 @@ from airflow.configuration import conf from airflow.models import Connection, DagModel, Pool, Variable from airflow.models.dagbundle import DagBundleModel +from airflow.models.revoked_token import RevokedToken from airflow.models.team import Team, dag_bundle_team_association_table from airflow.typing_compat import Unpack from airflow.utils.log.logging_mixin import LoggingMixin @@ -132,6 +133,10 @@ def deserialize_user(self, token: dict[str, Any]) -> T: def serialize_user(self, user: T) -> dict[str, Any]: """Create a subject and extra claims dict from a user object.""" + def revoke_token(self, token: str) -> None: + """Revoke a JWT token by persisting its JTI in the database.""" + self._get_token_validator().revoke_token(token) + async def get_user_from_token(self, token: str) -> BaseUser: """Verify the JWT token is valid and create a user object from it if valid.""" try: @@ -140,6 +145,9 @@ async def get_user_from_token(self, token: str) -> BaseUser: log.error("JWT token is not valid: %s", e) raise e + if (jti := payload.get("jti")) and RevokedToken.is_revoked(jti): + raise InvalidTokenError("Token has been revoked") + try: return self.deserialize_user(payload) except (ValueError, KeyError) as e: diff --git a/airflow-core/src/airflow/api_fastapi/auth/tokens.py b/airflow-core/src/airflow/api_fastapi/auth/tokens.py index 8fdaed0eb9e4c..e415cf101f8b9 100644 --- a/airflow-core/src/airflow/api_fastapi/auth/tokens.py +++ b/airflow-core/src/airflow/api_fastapi/auth/tokens.py @@ -35,6 +35,7 @@ from cryptography.hazmat.primitives.serialization import load_pem_private_key from airflow._shared.timezones import timezone +from airflow.models.revoked_token import RevokedToken if TYPE_CHECKING: from jwt.algorithms import AllowedKeys, AllowedPrivateKeys @@ -345,6 +346,15 @@ async def avalidated_claims( return claims + def revoke_token(self, token: str) -> None: + """Validate the token, extract jti and exp, and revoke it in the database.""" + try: + claims = self.validated_claims(token) + if (jti := claims.get("jti")) and (exp := claims.get("exp")): + RevokedToken.revoke(jti, datetime.fromtimestamp(exp, tz=timezone.utc)) + except (jwt.InvalidTokenError, Exception): + log.warning("Failed to revoke token", exc_info=True) + def status(self): if self.jwks: self.jwks.status() diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/auth.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/auth.py index 103b88d4a34ff..dcadd5123e610 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/auth.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/auth.py @@ -16,6 +16,7 @@ # under the License. from __future__ import annotations +import structlog from fastapi import HTTPException, Request, status from fastapi.responses import RedirectResponse @@ -25,6 +26,8 @@ from airflow.api_fastapi.core_api.security import AuthManagerDep, is_safe_url from airflow.configuration import conf +log = structlog.get_logger(logger_name=__name__) + auth_router = AirflowRouter(tags=["Login"], prefix="/auth") @@ -55,6 +58,10 @@ def logout(request: Request, auth_manager: AuthManagerDep) -> RedirectResponse: if logout_url: return RedirectResponse(logout_url) + # Revoke the current token before deleting the cookie + if token_str := request.cookies.get(COOKIE_NAME_JWT_TOKEN): + auth_manager.revoke_token(token_str) + secure = request.base_url.scheme == "https" or bool(conf.get("api", "ssl_cert", fallback="")) response = RedirectResponse(auth_manager.get_url_login()) response.delete_cookie( diff --git a/airflow-core/src/airflow/migrations/versions/0102_3_2_0_add_revoked_token_table.py b/airflow-core/src/airflow/migrations/versions/0102_3_2_0_add_revoked_token_table.py new file mode 100644 index 0000000000000..8a639d3f16506 --- /dev/null +++ b/airflow-core/src/airflow/migrations/versions/0102_3_2_0_add_revoked_token_table.py @@ -0,0 +1,54 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Add revoked_token table. + +Revision ID: 53ff648b8a26 +Revises: a5a3e5eb9b8d +Create Date: 2026-02-01 00:00:00.000000 + +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +from airflow.utils.sqlalchemy import UtcDateTime + +# revision identifiers, used by Alembic. +revision = "53ff648b8a26" +down_revision = "a5a3e5eb9b8d" +branch_labels = None +depends_on = None +airflow_version = "3.2.0" + + +def upgrade(): + """Add revoked_token table.""" + op.create_table( + "revoked_token", + sa.Column("jti", sa.String(32), primary_key=True, nullable=False), + sa.Column("exp", UtcDateTime, nullable=False, index=True), + ) + + +def downgrade(): + """Drop revoked_token table.""" + op.drop_table("revoked_token") diff --git a/airflow-core/src/airflow/models/__init__.py b/airflow-core/src/airflow/models/__init__.py index 3dc1c050bae0b..400151124e0df 100644 --- a/airflow-core/src/airflow/models/__init__.py +++ b/airflow-core/src/airflow/models/__init__.py @@ -69,6 +69,7 @@ def import_all_models(): import airflow.models.dagwarning import airflow.models.deadline_alert import airflow.models.errors + import airflow.models.revoked_token import airflow.models.serialized_dag import airflow.models.taskinstancehistory import airflow.models.tasklog diff --git a/airflow-core/src/airflow/models/revoked_token.py b/airflow-core/src/airflow/models/revoked_token.py new file mode 100644 index 0000000000000..39dc25c688454 --- /dev/null +++ b/airflow-core/src/airflow/models/revoked_token.py @@ -0,0 +1,79 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import time +from datetime import datetime, timezone +from typing import TYPE_CHECKING, ClassVar + +import structlog +from sqlalchemy import String, delete, exists, select +from sqlalchemy.orm import Mapped, mapped_column + +from airflow.configuration import conf +from airflow.models.base import Base +from airflow.utils.session import NEW_SESSION, provide_session +from airflow.utils.sqlalchemy import UtcDateTime + +if TYPE_CHECKING: + from sqlalchemy.orm import Session + +log = structlog.get_logger(__name__) + + +class RevokedToken(Base): + """Stores revoked JWT token JTIs to support token invalidation on logout.""" + + __tablename__ = "revoked_token" + + # Track last cleanup time to avoid running cleanup on every request + _last_cleanup_time: ClassVar[float] = 0.0 + + jti: Mapped[str] = mapped_column(String(32), primary_key=True) + exp: Mapped[datetime] = mapped_column(UtcDateTime, nullable=False, index=True) + + @classmethod + @provide_session + def revoke(cls, jti: str, exp: datetime, session: Session = NEW_SESSION) -> None: + """Add a token JTI to the revoked tokens.""" + session.merge(cls(jti=jti, exp=exp)) + + @classmethod + @provide_session + def is_revoked(cls, jti: str, session: Session = NEW_SESSION) -> bool: + """Check if a token JTI has been revoked.""" + cls._maybe_cleanup_expired(session) + return bool(session.scalar(select(exists().where(cls.jti == jti)))) + + @classmethod + def _maybe_cleanup_expired(cls, session: Session) -> None: + """ + Periodically clean up expired revoked tokens. + + Cleanup interval is based on jwt_expiration_time config to ensure expired + tokens are cleaned up after they're no longer useful. Uses monotonic time + to track intervals. + """ + now = time.monotonic() + cleanup_interval = conf.getint("api_auth", "jwt_expiration_time", fallback=3600) * 2 + if now - cls._last_cleanup_time >= cleanup_interval: + cls._last_cleanup_time = now + try: + session.execute(delete(cls).where(cls.exp < datetime.now(tz=timezone.utc))) + except Exception: + log.exception("Failed to clean up expired revoked tokens") diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index 2cd135e05d54f..2b3ea410c5f76 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -112,7 +112,7 @@ class MappedClassProtocol(Protocol): "3.0.0": "29ce7909c52b", "3.0.3": "fe199e1abd77", "3.1.0": "cc92b33c6709", - "3.2.0": "a5a3e5eb9b8d", + "3.2.0": "53ff648b8a26", } # Prefix used to identify tables holding data moved during migration. diff --git a/airflow-core/src/airflow/utils/db_cleanup.py b/airflow-core/src/airflow/utils/db_cleanup.py index 025303ce8f2f7..e6b5283669b86 100644 --- a/airflow-core/src/airflow/utils/db_cleanup.py +++ b/airflow-core/src/airflow/utils/db_cleanup.py @@ -171,6 +171,7 @@ def readable_config(self): keep_last_group_by=["dag_id"], ), _TableConfig(table_name="deadline", recency_column_name="deadline_time", dag_id_column_name="dag_id"), + _TableConfig(table_name="revoked_token", recency_column_name="exp"), ] # We need to have `fallback="database"` because this is executed at top level code and provider configuration diff --git a/airflow-core/tests/unit/api_fastapi/auth/managers/test_base_auth_manager.py b/airflow-core/tests/unit/api_fastapi/auth/managers/test_base_auth_manager.py index 71769ef49a600..22efbc426aea7 100644 --- a/airflow-core/tests/unit/api_fastapi/auth/managers/test_base_auth_manager.py +++ b/airflow-core/tests/unit/api_fastapi/auth/managers/test_base_auth_manager.py @@ -203,6 +203,29 @@ async def test_get_user_from_token(self, mock_deserialize_user, mock__get_token_ signer.avalidated_claims.assert_called_once_with(token) assert result == user + @patch( + "airflow.models.revoked_token.RevokedToken.is_revoked", + return_value=True, + ) + @patch( + "airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager._get_token_validator", + autospec=True, + ) + @pytest.mark.asyncio + async def test_get_user_from_token_revoked( + self, mock__get_token_validator, mock_is_revoked, auth_manager + ): + token = "token" + payload = {"jti": "some-jti"} + signer = AsyncMock(spec=JWTValidator) + signer.avalidated_claims.return_value = payload + mock__get_token_validator.return_value = signer + + with pytest.raises(InvalidTokenError, match="Token has been revoked"): + await auth_manager.get_user_from_token(token) + + mock_is_revoked.assert_called_once_with("some-jti") + @patch( "airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager._get_token_validator", autospec=True, @@ -224,6 +247,19 @@ async def test_get_user_from_token_invalid_token_payload( mock_deserialize_user.assert_called_once_with(payload) signer.avalidated_claims.assert_called_once_with(token) + @patch( + "airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager._get_token_validator", + autospec=True, + ) + def test_revoke_token(self, mock__get_token_validator, auth_manager): + token = "token" + validator = Mock(spec=JWTValidator) + mock__get_token_validator.return_value = validator + + auth_manager.revoke_token(token) + + validator.revoke_token.assert_called_once_with(token) + @patch("airflow.api_fastapi.auth.managers.base_auth_manager.JWTGenerator", autospec=True) @patch.object(EmptyAuthManager, "serialize_user") def test_generate_jwt_token(self, mock_serialize_user, mock_jwt_generator, auth_manager): diff --git a/airflow-core/tests/unit/api_fastapi/auth/test_tokens.py b/airflow-core/tests/unit/api_fastapi/auth/test_tokens.py index 34ac077f8e220..d0fe41e14500a 100644 --- a/airflow-core/tests/unit/api_fastapi/auth/test_tokens.py +++ b/airflow-core/tests/unit/api_fastapi/auth/test_tokens.py @@ -239,6 +239,88 @@ async def test_jwt_generate_validate_roundtrip_with_jwks(private_key, algorithm, assert await validator.avalidated_claims(token) +class TestRevokeToken: + pytestmark = [pytest.mark.db_test] + + @pytest.fixture(autouse=True) + def cleanup_revoked_tokens(self): + from tests_common.test_utils.db import clear_db_revoked_tokens + + clear_db_revoked_tokens() + yield + clear_db_revoked_tokens() + + def test_revoke_token_persists_in_db(self): + """Test that revoke_token validates the token and persists the jti in the database.""" + import time + + from airflow.models.revoked_token import RevokedToken + + now = int(time.time()) + payload = { + "sub": "user", + "jti": "revoke-test-jti", + "exp": now + 3600, + "iat": now, + "nbf": now, + "aud": "test", + } + token = jwt.encode(payload, "secret", algorithm="HS256") + + validator = JWTValidator(secret_key="secret", audience="test", algorithm=["HS256"], leeway=0) + validator.revoke_token(token) + + assert RevokedToken.is_revoked("revoke-test-jti") is True + + def test_revoke_token_without_jti_does_not_persist(self): + """Test that a token without jti does not create a revoked token entry.""" + import time + + from airflow.models.revoked_token import RevokedToken + + now = int(time.time()) + payload = {"sub": "user", "exp": now + 3600, "iat": now, "nbf": now, "aud": "test"} + token = jwt.encode(payload, "secret", algorithm="HS256") + + validator = JWTValidator(secret_key="secret", audience="test", algorithm=["HS256"], leeway=0) + validator.revoke_token(token) + + assert RevokedToken.is_revoked("any-jti") is False + + def test_revoke_token_with_invalid_token_does_not_raise(self): + """Test that revoke_token logs a warning instead of raising for an invalid token.""" + from airflow.models.revoked_token import RevokedToken + + validator = JWTValidator(secret_key="secret", audience="test", algorithm=["HS256"], leeway=0) + validator.revoke_token("invalid-token") + + assert RevokedToken.is_revoked("any-jti") is False + + def test_revoke_token_with_db_error_does_not_raise(self): + """Test that revoke_token handles database errors gracefully.""" + import time + from unittest.mock import patch + + from sqlalchemy.exc import SQLAlchemyError + + now = int(time.time()) + payload = { + "sub": "user", + "jti": "db-error-jti", + "exp": now + 3600, + "iat": now, + "nbf": now, + "aud": "test", + } + token = jwt.encode(payload, "secret", algorithm="HS256") + + validator = JWTValidator(secret_key="secret", audience="test", algorithm=["HS256"], leeway=0) + with patch( + "airflow.models.revoked_token.RevokedToken.revoke", side_effect=SQLAlchemyError("db down") + ): + validator.revoke_token(token) + + @pytest.fixture(scope="session") def rsa_private_key(): return generate_private_key() diff --git a/airflow-core/tests/unit/api_fastapi/conftest.py b/airflow-core/tests/unit/api_fastapi/conftest.py index 36b46f8e51812..6d57f90a77eb5 100644 --- a/airflow-core/tests/unit/api_fastapi/conftest.py +++ b/airflow-core/tests/unit/api_fastapi/conftest.py @@ -76,9 +76,12 @@ def test_client(request): ).generate( auth_manager.serialize_user(SimpleAuthManagerUser(username="test", role="admin")), ) - yield TestClient( - app, headers={"Authorization": f"Bearer {token}"}, base_url=f"{BASE_URL}{get_api_path(request)}" - ) + with mock.patch("airflow.models.revoked_token.RevokedToken.is_revoked", return_value=False): + yield TestClient( + app, + headers={"Authorization": f"Bearer {token}"}, + base_url=f"{BASE_URL}{get_api_path(request)}", + ) @pytest.fixture @@ -101,9 +104,12 @@ def unauthorized_test_client(request): token = auth_manager._get_token_signer().generate( auth_manager.serialize_user(SimpleAuthManagerUser(username="dummy", role=None)) ) - yield TestClient( - app, headers={"Authorization": f"Bearer {token}"}, base_url=f"{BASE_URL}{get_api_path(request)}" - ) + with mock.patch("airflow.models.revoked_token.RevokedToken.is_revoked", return_value=False): + yield TestClient( + app, + headers={"Authorization": f"Bearer {token}"}, + base_url=f"{BASE_URL}{get_api_path(request)}", + ) @pytest.fixture diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_auth.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_auth.py index 548b16eb8be0f..db04f91e34c85 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_auth.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_auth.py @@ -16,13 +16,17 @@ # under the License. from __future__ import annotations +import time from unittest.mock import MagicMock, patch +import jwt import pytest from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN +from airflow.models.revoked_token import RevokedToken from tests_common.test_utils.config import conf_vars +from tests_common.test_utils.db import clear_db_revoked_tokens AUTH_MANAGER_LOGIN_URL = "http://some_login_url" AUTH_MANAGER_LOGOUT_URL = "http://some_logout_url" @@ -75,6 +79,12 @@ def test_should_respond_400(self, test_client, params): class TestLogout(TestAuthEndpoint): + @pytest.fixture(autouse=True) + def cleanup_revoked_tokens(self): + clear_db_revoked_tokens() + yield + clear_db_revoked_tokens() + @pytest.mark.parametrize( ("mock_logout_url", "expected_redirection", "delete_cookies"), [ @@ -100,3 +110,74 @@ def test_should_respond_307( if delete_cookies: cookies = response.headers.get_list("set-cookie") assert any(f"{COOKIE_NAME_JWT_TOKEN}=" in c for c in cookies) + + def test_logout_with_invalid_token_does_not_raise(self, test_client): + """Test that logout with an invalid token does not raise.""" + test_client.app.state.auth_manager.get_url_logout.return_value = None + + test_client.cookies.set(COOKIE_NAME_JWT_TOKEN, "not-a-valid-jwt") + + response = test_client.get("/auth/logout", follow_redirects=False) + + assert response.status_code == 307 + + +class TestLogoutTokenRevocation: + """Tests for token revocation on logout, using real DB queries without mocks.""" + + pytestmark = [pytest.mark.db_test] + + @pytest.fixture(autouse=True) + def cleanup_revoked_tokens(self): + clear_db_revoked_tokens() + yield + clear_db_revoked_tokens() + + @pytest.fixture + def logout_client(self): + """A test client without the is_revoked mock so revocation tests hit the real DB.""" + from fastapi.testclient import TestClient + + from airflow.api_fastapi.app import create_app + + with conf_vars( + { + ( + "core", + "auth_manager", + ): "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager" + } + ): + app = create_app() + yield TestClient(app, base_url="http://testserver/api/v2") + + def test_logout_revokes_token(self, logout_client): + """Test that logout revokes the JWT token and persists it in the database.""" + now = int(time.time()) + token_payload = { + "sub": "admin", + "jti": "test-jti-123", + "exp": now + 3600, + "iat": now, + "nbf": now, + "aud": "apache-airflow", + } + auth_manager = logout_client.app.state.auth_manager + signer = auth_manager._get_token_signer() + token_str = jwt.encode(token_payload, signer._secret_key, algorithm=signer.algorithm) + + logout_client.cookies.set(COOKIE_NAME_JWT_TOKEN, token_str) + with patch.object(auth_manager, "get_url_logout", return_value=None): + response = logout_client.get("/auth/logout", follow_redirects=False) + + assert response.status_code == 307 + assert RevokedToken.is_revoked("test-jti-123") is True + + def test_logout_without_cookie_does_not_revoke(self, logout_client): + """Test that logout without a cookie does not attempt to revoke.""" + auth_manager = logout_client.app.state.auth_manager + with patch.object(auth_manager, "get_url_logout", return_value=None): + response = logout_client.get("/auth/logout", follow_redirects=False) + + assert response.status_code == 307 + assert RevokedToken.is_revoked("nonexistent-jti") is False diff --git a/airflow-core/tests/unit/models/test_revoked_token.py b/airflow-core/tests/unit/models/test_revoked_token.py new file mode 100644 index 0000000000000..25f904b2ab797 --- /dev/null +++ b/airflow-core/tests/unit/models/test_revoked_token.py @@ -0,0 +1,92 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +from airflow.models.revoked_token import RevokedToken + + +class TestRevokedTokenModel: + def test_revoke_inserts_row(self): + """Test that revoke calls session.merge with a RevokedToken instance.""" + mock_session = MagicMock() + exp = datetime.now(tz=timezone.utc) + timedelta(hours=1) + RevokedToken.revoke("test-jti-123", exp, session=mock_session) + mock_session.merge.assert_called_once() + arg = mock_session.merge.call_args[0][0] + assert isinstance(arg, RevokedToken) + assert arg.jti == "test-jti-123" + assert arg.exp == exp + + def test_is_revoked_returns_true(self): + """Test that a revoked JTI is detected.""" + mock_session = MagicMock() + mock_session.scalar.return_value = True + result = RevokedToken.is_revoked("known-jti", session=mock_session) + assert result is True + + def test_is_revoked_returns_false(self): + """Test that an unknown JTI returns False.""" + mock_session = MagicMock() + mock_session.scalar.return_value = False + result = RevokedToken.is_revoked("unknown-jti", session=mock_session) + assert result is False + + +class TestRevokedTokenCleanup: + """Tests for automatic periodic cleanup of expired revoked tokens.""" + + def test_cleanup_runs_when_interval_passed(self): + """Cleanup should run when enough time has passed since last cleanup.""" + mock_session = MagicMock() + mock_session.scalar.return_value = False + + original_last_cleanup = RevokedToken._last_cleanup_time + try: + RevokedToken._last_cleanup_time = 0.0 + with ( + patch("airflow.models.revoked_token.time.monotonic", return_value=8000.0), + patch("airflow.models.revoked_token.conf.getint", return_value=3600), + ): + RevokedToken.is_revoked("test-jti", session=mock_session) + + # session.execute should be called for DELETE + mock_session.execute.assert_called_once() + finally: + RevokedToken._last_cleanup_time = original_last_cleanup + + def test_cleanup_skips_when_interval_not_passed(self): + """Cleanup should skip when not enough time has passed.""" + mock_session = MagicMock() + mock_session.scalar.return_value = False + + original_last_cleanup = RevokedToken._last_cleanup_time + try: + RevokedToken._last_cleanup_time = 4000.0 + # cleanup_interval = 3600 * 2 = 7200, so 4500 - 4000 = 500 < 7200 skips cleanup + with ( + patch("airflow.models.revoked_token.time.monotonic", return_value=4500.0), + patch("airflow.models.revoked_token.conf.getint", return_value=3600), + ): + RevokedToken.is_revoked("test-jti", session=mock_session) + + # session.execute should NOT be called + mock_session.execute.assert_not_called() + finally: + RevokedToken._last_cleanup_time = original_last_cleanup diff --git a/devel-common/src/tests_common/test_utils/db.py b/devel-common/src/tests_common/test_utils/db.py index d29cc06141be9..acaa24df35b3f 100644 --- a/devel-common/src/tests_common/test_utils/db.py +++ b/devel-common/src/tests_common/test_utils/db.py @@ -427,6 +427,13 @@ def clear_db_teams(): session.execute(delete(Team)) +def clear_db_revoked_tokens(): + with create_session() as session: + from airflow.models.revoked_token import RevokedToken + + session.execute(delete(RevokedToken)) + + def clear_dag_specific_permissions(): if "FabAuthManager" not in conf.get("core", "auth_manager"): return