SFTP sensors can make incorrect newer_than comparisons when timezones differ

[mifuha]

Under which category would you file this issue?

Providers

Apache Airflow version

3.4.0 (development checkout at main commit 224d34121c877c21e8dee97520b05c02f578a839, not a released-version test).

What happened and how to reproduce it?

SFTP sensors can make the wrong decision about newer_than when the process timezone and Airflow's default timezone differ. A file can be accepted even though it is too old, or rejected even though it is new enough.

I reproduced this in the single-file SFTPTrigger path and in the synchronous SFTPSensor, both with and without a file pattern. The reproduction uses the real hook conversions and sensor/trigger code, with SFTP I/O mocked.

The file timestamp is 1704110400, which is 2024-01-01 12:00 UTC. All thresholds below are on that date and include a UTC offset. All three paths gave these results:

• Process timezone UTC, Airflow default America/New_York:
  Threshold 14:00 UTC. Expected: reject. Actual: accept.

• Process timezone America/New_York, Airflow default UTC:
  Threshold 10:00 UTC. Expected: accept. Actual: reject.

• Both timezones UTC:
  Threshold 10:00 UTC. Expected: accept. Actual: accept.

• Both timezones UTC:
  Threshold 14:00 UTC. Expected: reject. Actual: reject.

Reproduction

Tested at main commit 224d34121c877c21e8dee97520b05c02f578a839, using its locked dependencies. Save the script below as dev/sftp_timezone_probe.py in that checkout, then run:

uv sync --frozen --project providers/sftp --python 3.12
export AIRFLOW_HOME=/tmp/airflow-sftp-probe-home
export _AIRFLOW_PROCESS_CONTEXT=client
TZ=UTC AIRFLOW__CORE__DEFAULT_TIMEZONE=America/New_York uv run --frozen --project providers/sftp python dev/sftp_timezone_probe.py
TZ=America/New_York AIRFLOW__CORE__DEFAULT_TIMEZONE=UTC uv run --frozen --project providers/sftp python dev/sftp_timezone_probe.py
TZ=UTC AIRFLOW__CORE__DEFAULT_TIMEZONE=UTC uv run --frozen --project providers/sftp python dev/sftp_timezone_probe.py

Each command starts a fresh process with normal Airflow initialization. The script prints the effective timezones and the expected and actual decisions. Exit code zero means the script completed, not that the decisions were correct.

For the trigger, the script stops after the first unsuccessful poll by raising asyncio.CancelledError at the polling sleep. No SFTP server or credentials are needed.

Reproduction script:

# 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.
# /// script
# requires-python = ">=3.10"
# ///
from __future__ import annotations

import asyncio
import calendar
import datetime
import json
import os
import platform
import stat
import time
from importlib.metadata import version
from unittest import mock

import asyncssh
import paramiko

from airflow import settings
from airflow.providers.common.compat.sdk import Connection, timezone
from airflow.providers.sftp.hooks.sftp import SFTPHook, SFTPHookAsync
from airflow.providers.sftp.sensors.sftp import SFTPSensor
from airflow.providers.sftp.triggers.sftp import SFTPTrigger

MTIME = 1704110400


@mock.patch(
    "airflow.providers.sftp.triggers.sftp.asyncio.sleep", autospec=True, side_effect=asyncio.CancelledError
)
@mock.patch.object(SFTPHookAsync, "_get_conn", autospec=True)
async def inspect_async_path(get_conn, sleep, threshold):
    client = mock.AsyncMock(spec=asyncssh.SFTPClient)
    client.stat.return_value = asyncssh.SFTPAttrs(mtime=MTIME)
    sftp_cm = mock.MagicMock()
    sftp_cm.__aenter__ = mock.AsyncMock(return_value=client)
    ssh = mock.MagicMock(spec=asyncssh.SSHClientConnection)
    ssh.__aenter__.return_value = ssh
    ssh.start_sftp_client.return_value = sftp_cm
    get_conn.return_value = ssh
    mod_time = await SFTPHookAsync().get_mod_time("/files/file.txt")
    trigger = SFTPTrigger(path="/files/file.txt", newer_than=threshold)
    generator = trigger.run()
    try:
        try:
            event = await anext(generator)
        except asyncio.CancelledError:
            event = None
        if event is None:
            sleep.assert_awaited_once_with(trigger.poke_interval)
        else:
            if event.payload["status"] != "success":
                raise AssertionError(event)
        if client.stat.await_count != 2 or ssh.__aexit__.await_count != 2:
            raise AssertionError("Expected two stat calls and two closed connection contexts")
        return {"mod_time": mod_time, "accepted": event is not None}
    finally:
        await generator.aclose()


@mock.patch.object(SFTPHook, "get_connection", spec=True)
@mock.patch.object(SFTPHook, "get_managed_conn", autospec=True)
def inspect_sync_paths(managed_conn, get_connection, threshold):
    get_connection.return_value = Connection(conn_id="sftp_default", conn_type="sftp", host="mock.invalid")
    attrs = paramiko.SFTPAttributes()
    attrs.st_mtime = MTIME
    attrs.st_mode = stat.S_IFREG | 0o644
    attrs.filename = "file.txt"
    client = mock.MagicMock(spec=paramiko.SFTPClient)
    client.stat.return_value = attrs
    client.listdir_attr.return_value = [attrs]
    managed_conn.return_value.__enter__.return_value = client
    hook = SFTPHook()
    result = {"mod_time": hook.get_mod_time("/files/file.txt")}
    for pattern in ("", "*.txt"):
        sensor = SFTPSensor(
            task_id="mtime_probe",
            path="/files" if pattern else "/files/file.txt",
            file_pattern=pattern,
            newer_than=threshold,
        )
        sensor.hook = hook
        result["pattern" if pattern else "single_file"] = {
            "get_files": sensor._get_files(),
            "poke": sensor.poke({}),
        }
    return result


def main():
    time.tzset()
    if calendar.timegm((2024, 1, 1, 12, 0, 0)) != MTIME:
        raise AssertionError("Unexpected epoch timestamp")
    packages = [
        "apache-airflow",
        "apache-airflow-core",
        "apache-airflow-task-sdk",
        "apache-airflow-providers-sftp",
        "apache-airflow-providers-ssh",
        "apache-airflow-providers-common-compat",
        "asyncssh",
        "paramiko",
        "pendulum",
        "python-dateutil",
        "pytest",
        "pytest-asyncio",
        "tzdata",
    ]
    result = {
        "python": platform.python_version(),
        "versions": {package: version(package) for package in packages},
        "process_tz": os.environ.get("TZ"),
        "process_time": time.strftime("%Y-%m-%dT%H:%M:%S%z", time.localtime(MTIME)),
        "core_timezone": str(settings.TIMEZONE),
        "provider_timezone": str(timezone.datetime(2024, 1, 1).tzinfo),
        "helper_module": timezone.convert_to_utc.__module__,
        "helper_file": __import__(timezone.convert_to_utc.__module__, fromlist=[""]).__file__,
        "source": SFTPTrigger.run.__code__.co_filename,
        "library_optout": os.environ.get("_AIRFLOW__AS_LIBRARY"),
        "unit_test_mode": os.environ.get("AIRFLOW__CORE__UNIT_TEST_MODE"),
        "process_context": os.environ.get("_AIRFLOW_PROCESS_CONTEXT"),
        "cases": [],
    }
    for hour in (10, 14):
        threshold = datetime.datetime(2024, 1, 1, hour, tzinfo=datetime.timezone.utc)
        result["cases"].append(
            {
                "threshold": threshold.isoformat(),
                "expected": hour == 10,
                "sync": inspect_sync_paths(threshold=threshold),
                "async_single_file": asyncio.run(inspect_async_path(threshold=threshold)),
            }
        )
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

What you think should happen instead?

The same file timestamp and timezone-aware threshold should give the same result regardless of these timezone settings. A file whose modification time equals the threshold should still be accepted.

Cause

Both get_mod_time() methods turn the timestamp into a local-time string without an offset. The sensor/trigger then interprets that string using Airflow's default timezone.

For example, with the process timezone set to UTC and Airflow's default set to America/New_York, 12:00 UTC becomes a string representing 12:00. That is then interpreted as 12:00 New York time, or 17:00 UTC on this date.

Relevant source: hook conversions, sensor comparison, trigger comparison.

Operating System

Ubuntu 24.04.4 under WSL2; Python 3.12.3.

Deployment

Virtualenv installation from source with locked dependencies. No live Airflow deployment was tested.

Apache Airflow Provider(s)

sftp

Versions of Apache Airflow Providers

• apache-airflow-providers-sftp: 6.0.1
• apache-airflow-providers-ssh: 6.0.1
• apache-airflow-providers-common-compat: 1.19.0

Other relevant versions: Task SDK 1.4.0, AsyncSSH 2.24.0, Paramiko 5.0.0, Pendulum 3.2.0.

Official Helm Chart version

Not Applicable

Kubernetes Version

Not Applicable

Helm Chart configuration

Not Applicable

Docker Image customizations

Not Applicable

Anything else?

The comparison needs a raw timestamp or a timezone-aware datetime, while keeping compatibility with the existing get_mod_time() string APIs. Changing only the strings to UTC would still leave callers interpreting them in Airflow's default timezone.

I also tested a local patch for the separate pattern-based trigger branch, where the raw timestamp is already available. That patch converts the timestamp directly to an aware UTC datetime. It does not change the three paths reported here, and all three still reproduced the results above with that patch applied.

I checked open and closed issues and PRs and did not find an equivalent report or fix. #65442 changes timezone imports; #73312 changes callback behavior.

Are you willing to submit PR?

Yes. I have a tested local patch for the pattern-based trigger path. The single-file and synchronous paths described here still need a compatible fix.

Code of Conduct

• I agree to follow this project's Code of Conduct.

— @mifuha

AI assistance: I used Codex to help investigate this issue and prepare the reproduction script.

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.