Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/logging/decorators.py: 87%
106 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 14:22 +0000
1# Licensed to the Apache Software Foundation (ASF) under one
2# or more contributor license agreements. See the NOTICE file
3# distributed with this work for additional information
4# regarding copyright ownership. The ASF licenses this file
5# to you under the Apache License, Version 2.0 (the
6# "License"); you may not use this file except in compliance
7# with the License. You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing,
12# software distributed under the License is distributed on an
13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14# KIND, either express or implied. See the License for the
15# specific language governing permissions and limitations
16# under the License.
17from __future__ import annotations
19import itertools
20import json
21import logging
22from datetime import datetime
24import pendulum
25from fastapi import Request
26from pendulum.parsing.exceptions import ParserError
28from airflow._shared.secrets_masker import secrets_masker
29from airflow.api_fastapi.common.db.common import SessionDep
30from airflow.api_fastapi.core_api.security import GetUserDep
31from airflow.models import Log
33logger = logging.getLogger(__name__)
36def _sanitize_for_stdlib_log(value: str) -> str:
37 """
38 Strip CR/LF from a user-supplied value before passing it to stdlib's ``%s``-style logging.
40 Defends against log injection when the deployment is configured with a non-JSON
41 (plain-text) log formatter: a newline in the value would otherwise let an attacker forge
42 log lines. ``structlog``-style formatters are unaffected, but the access-log path uses
43 the stdlib logger here, so the sanitisation is unconditional.
44 """
45 return value.replace("\r", " ").replace("\n", " ")
48def _mask_bulk_entities(extra_fields, mask_entity):
49 """
50 Apply per-entity masking to a bulk request body.
52 A ``BulkBody`` has exactly one top-level field, ``actions``; the entities carrying the
53 secrets sit two levels down, in ``actions[].entities[]``. The per-entity maskers below
54 inspect top-level key names, so handing them a bulk body means they see only the key
55 ``actions`` and pass its whole payload through untouched. Reach the entities first.
57 Returns ``None`` when the body is not bulk-shaped, so callers fall back to flat masking.
58 """
59 actions = extra_fields.get("actions")
60 if not isinstance(actions, list):
61 return None
63 masked_actions = []
64 for action in actions:
65 if not isinstance(action, dict):
66 masked_actions.append(action)
67 continue
68 entities = action.get("entities")
69 if not isinstance(entities, list):
70 masked_actions.append(action)
71 continue
72 # ``delete`` actions may list bare id/key strings rather than entity objects;
73 # those carry no secret and are left as they are.
74 masked_actions.append(
75 {
76 **action,
77 "entities": [mask_entity(e) if isinstance(e, dict) else e for e in entities],
78 }
79 )
80 return {**extra_fields, "actions": masked_actions}
83def _mask_connection_fields(extra_fields):
84 """Mask connection fields, for either a single-entity or a bulk request body."""
85 bulk = _mask_bulk_entities(extra_fields, _mask_connection_entity)
86 if bulk is not None:
87 return bulk
88 return _mask_connection_entity(extra_fields)
91def _mask_connection_entity(extra_fields):
92 """Mask the fields of one connection."""
93 result = {}
94 for k, v in extra_fields.items():
95 if k == "extra" and v:
96 try:
97 parsed_extra = json.loads(v)
98 if isinstance(parsed_extra, dict):
99 # Connection ``extra`` can carry values under arbitrary key names, so the
100 # audit-log entry records only *which* ``extra`` fields were present and masks
101 # every value rather than deciding what to mask from the key name.
102 result[k] = {ek: "***" for ek in parsed_extra}
103 else:
104 result[k] = "Expected JSON object in `extra` field, got non-dict JSON"
105 except (json.JSONDecodeError, TypeError):
106 # ``extra`` is declared as a string, but this runs on the raw body before
107 # validation, so it can arrive as any JSON type -- a number or an already-decoded
108 # object makes ``json.loads`` raise TypeError rather than JSONDecodeError. Both
109 # are recorded without the value, instead of raising out of the audit-log path.
110 result[k] = "Encountered non-JSON in `extra` field"
111 else:
112 result[k] = secrets_masker.redact(v, k)
113 return result
116def _mask_variable_fields(extra_fields):
117 """Mask variable values, for either a single-entity or a bulk request body."""
118 bulk = _mask_bulk_entities(extra_fields, _mask_variable_entity)
119 if bulk is not None:
120 return bulk
121 return _mask_variable_entity(extra_fields)
124def _mask_variable_entity(extra_fields):
125 """
126 Mask the variable value.
128 The variable requests values and args comes in this form:
129 {'key': 'key_content', 'val': 'val_content', 'description': 'description_content'}
131 The value is masked unconditionally — the audit log records that a variable
132 changed, not its contents, so a secret stored under any key name (not just a
133 sensitive-looking one) is never persisted to the log.
134 """
135 result = {}
136 for k, v in extra_fields.items():
137 result[k] = "***" if k in ("val", "value") else v
138 return result
141def action_logging(event: str | None = None):
142 async def log_action(
143 request: Request,
144 session: SessionDep,
145 user: GetUserDep,
146 ):
147 """Log user actions."""
148 event_name = event or request.scope["endpoint"].__name__
149 skip_dry_run_events = {"clear_dag_run", "post_clear_task_instances"}
151 if not user: 151 ↛ 152line 151 didn't jump to line 152 because the condition on line 151 was never true
152 user_name = "anonymous"
153 user_display = ""
154 else:
155 user_name = user.get_name()
156 user_display = user.get_name()
158 has_json_body = "application/json" in request.headers.get("content-type", "") and await request.body()
159 request_body = {}
160 masked_body_json = {}
162 if has_json_body:
163 # Non-dict bodies fall through to the endpoint's own 422.
164 parsed_body = await request.json()
165 if isinstance(parsed_body, dict):
166 request_body = parsed_body
167 masked_body_json = {k: secrets_masker.redact(v, k) for k, v in request_body.items()}
169 if event_name in skip_dry_run_events and request_body.get("dry_run", True):
170 return
172 fields_skip_logging = {
173 "csrf_token",
174 "_csrf_token",
175 "is_paused",
176 "dag_id",
177 "task_id",
178 "dag_run_id",
179 "run_id",
180 "logical_date",
181 }
183 extra_fields = {
184 k: secrets_masker.redact(v, k)
185 for k, v in itertools.chain(request.query_params.items(), request.path_params.items())
186 if k not in fields_skip_logging
187 }
188 if "variable" in event_name:
189 extra_fields = _mask_variable_fields(
190 {k: v for k, v in request_body.items()} if has_json_body else extra_fields
191 )
192 elif "connection" in event_name:
193 extra_fields = _mask_connection_fields(
194 {k: v for k, v in request_body.items()} if has_json_body else extra_fields
195 )
196 elif has_json_body:
197 extra_fields = {**extra_fields, **masked_body_json}
199 params = {
200 **request.query_params,
201 **request.path_params,
202 }
204 if has_json_body:
205 params.update(masked_body_json)
206 if params and "is_paused" in params:
207 extra_fields["is_paused"] = params["is_paused"]
209 extra_fields["method"] = request.method
211 # Create log entry
212 log = Log(
213 event=event_name,
214 task_instance=None,
215 owner=user_name,
216 owner_display_name=user_display,
217 extra=json.dumps(extra_fields),
218 task_id=params.get("task_id"),
219 dag_id=params.get("dag_id"),
220 run_id=params.get("run_id") or params.get("dag_run_id"),
221 )
223 if "logical_date" in request.query_params: 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true
224 logical_date_value = request.query_params.get("logical_date")
225 if logical_date_value:
226 try:
227 logical_date = pendulum.parse(logical_date_value, strict=False)
228 if not isinstance(logical_date, datetime):
229 raise ParserError
230 log.logical_date = logical_date
231 except ParserError:
232 logger.exception(
233 "Failed to parse logical_date from the request: %s",
234 _sanitize_for_stdlib_log(logical_date_value),
235 )
236 else:
237 logger.warning("Logical date is missing or empty")
238 session.add(log)
239 # Explicit commit to persist the access log independently if the path operation fails or not.
240 # Also it cannot be deferred to a 'function' scoped dependency because of the `request` parameter.
241 session.commit()
243 return log_action