Remote operations

SSRF guard for outbound HTTP requests.

validate_http_url rejects non-http(s) schemes, resolves the host, and rejects private / loopback / link-local / reserved IP ranges. Every remote function that accepts a user-supplied URL must pass it through here first.

automation_file.remote.url_validator.validate_http_url(url, *, allow_private=False)[source]

Return url if safe; raise UrlValidationException otherwise.

allow_private=True relaxes the private/loopback/link-local checks for callers that need to reach LAN services (e.g. on-prem WebDAV). Scheme and host checks still apply. Callers must opt in explicitly — the default remains strict SSRF blocking.

Parameters:
Return type:

str

SSRF-guarded HTTP downloader.

automation_file.remote.http_download.download_file(file_url, file_name, chunk_size=65536, timeout=15, max_bytes=20971520, progress_name=None, resume=False, expected_sha256=None)[source]

Download file_url to file_name with progress display.

Validates the URL against SSRF rules, disables redirects, enforces a size cap, retries transient network errors up to three times, and uses default TLS verification.

Pass progress_name to register the transfer with progress_registry so it can be polled or cancelled from the GUI. When resume=True the download streams into <file_name>.part and, if that file already exists from a previous interrupted run, sends Range: bytes=<existing>- to continue where it stopped (append mode); the .part file is atomically renamed to file_name on success. Pass expected_sha256 to verify the finished file; a mismatch is logged, the download is removed, and the function returns False. Returns True on success.

Parameters:

  • file_url (str)

  • file_name (str)

  • chunk_size (int)

  • timeout (int)

  • max_bytes (int)

  • progress_name (str | None)

  • resume (bool)

  • expected_sha256 (str | None)

Return type:

bool

Google Drive

Google Drive client (Singleton Facade).

Wraps OAuth2 credential loading and exposes a lazily-built service attribute that every operation module calls through.

class automation_file.remote.google_drive.client.GoogleDriveClient(scopes=('https://www.googleapis.com/auth/drive',))[source]

Bases: object

Holds credentials and the Drive API service handle.

Parameters:

scopes (tuple[str, ...])

later_init(token_path, credentials_path)[source]

Load / refresh credentials and build the Drive service.

Writes the refreshed token back to token_path with UTF-8 encoding.

Parameters:

  • token_path (str)

  • credentials_path (str)

Return type:

Any

require_service()[source]

Return self.service or raise if the client has not been initialised.

Return type:

Any

Delete-side Google Drive operations.

automation_file.remote.google_drive.delete_ops.drive_delete_file(file_id)[source]

Delete a file by Drive ID. Returns the API response or None.

Parameters:

file_id (str)

Return type:

Any | None

Folder (mkdir-equivalent) operations on Google Drive.

automation_file.remote.google_drive.folder_ops.drive_add_folder(folder_name)[source]

Create a folder on Drive. Returns the new folder’s ID or None.

Parameters:

folder_name (str)

Return type:

str | None

Search-side Google Drive operations.

automation_file.remote.google_drive.search_ops.drive_search_all_file()[source]

Return {name: id} for every file visible to the current token.

Return type:

dict[str, str] | None

automation_file.remote.google_drive.search_ops.drive_search_field(field_pattern)[source]

Return {name: id} for a list call with a custom fields= pattern.

Parameters:

field_pattern (str)

Return type:

dict[str, str] | None

automation_file.remote.google_drive.search_ops.drive_search_file_mimetype(mime_type)[source]

Return {name: id} for files matching mime_type (all pages).

Parameters:

mime_type (str)

Return type:

dict[str, str] | None

Permission / share operations on Google Drive.

automation_file.remote.google_drive.share_ops.drive_share_file_to_anyone(file_id, share_role='reader')[source]
Parameters:

  • file_id (str)

  • share_role (str)

Return type:

dict | None

automation_file.remote.google_drive.share_ops.drive_share_file_to_domain(file_id, domain, domain_role='reader')[source]
Parameters:

  • file_id (str)

  • domain (str)

  • domain_role (str)

Return type:

dict | None

automation_file.remote.google_drive.share_ops.drive_share_file_to_user(file_id, user, user_role='writer')[source]
Parameters:
Return type:

dict | None

Upload-side Google Drive operations.

automation_file.remote.google_drive.upload_ops.drive_upload_dir_to_drive(dir_path)[source]

Upload every file in dir_path (non-recursive) to the Drive root.

Parameters:

dir_path (str)

Return type:

list[dict | None]

automation_file.remote.google_drive.upload_ops.drive_upload_dir_to_folder(folder_id, dir_path)[source]

Upload every file in dir_path (non-recursive) to a Drive folder.

Parameters:

  • folder_id (str)

  • dir_path (str)

Return type:

list[dict | None]

automation_file.remote.google_drive.upload_ops.drive_upload_to_drive(file_path, file_name=None)[source]

Upload a single file to the Drive root.

Parameters:

  • file_path (str)

  • file_name (str | None)

Return type:

dict | None

automation_file.remote.google_drive.upload_ops.drive_upload_to_folder(folder_id, file_path, file_name=None)[source]

Upload a single file into a specific Drive folder.

Parameters:

  • folder_id (str)

  • file_path (str)

  • file_name (str | None)

Return type:

dict | None

Download-side Google Drive operations.

automation_file.remote.google_drive.download_ops.drive_download_file(file_id, file_name)[source]

Download a single file by ID to file_name on disk.

Returns the in-memory buffer on success, or None on failure. The file is only written after the download completes cleanly, so a failed request cannot leave an empty file behind.

Parameters:

  • file_id (str)

  • file_name (str)

Return type:

BytesIO | None

automation_file.remote.google_drive.download_ops.drive_download_file_from_folder(folder_name)[source]

Download every file inside the Drive folder named folder_name.

Parameters:

folder_name (str)

Return type:

dict[str, str] | None

S3

Bundled with automation_file; registered automatically by automation_file.core.action_registry.build_default_registry().

S3 delete operations.

automation_file.remote.s3.delete_ops.s3_delete_object(bucket, key)[source]

Delete s3://bucket/key. Returns True on success.

Parameters:
Return type:

bool

Azure Blob

Bundled with automation_file; registered automatically by automation_file.core.action_registry.build_default_registry().

Azure Blob delete operations.

automation_file.remote.azure_blob.delete_ops.azure_blob_delete_blob(container, blob_name)[source]

Delete a blob. Returns True on success.

Parameters:

  • container (str)

  • blob_name (str)

Return type:

bool

Dropbox

Bundled with automation_file; registered automatically by automation_file.core.action_registry.build_default_registry().

Dropbox delete operations.

automation_file.remote.dropbox_api.delete_ops.dropbox_delete_path(remote_path)[source]

Delete a file or folder at remote_path.

Parameters:

remote_path (str)

Return type:

bool

SFTP

Bundled with automation_file; registered automatically by automation_file.core.action_registry.build_default_registry(). Uses paramiko.RejectPolicy — unknown hosts are never auto-added.

SFTP delete operations.

automation_file.remote.sftp.delete_ops.sftp_delete_path(remote_path)[source]

Delete a remote file. (Directories require a recursive helper.)

Parameters:

remote_path (str)

Return type:

bool

FTP / FTPS

Bundled with automation_file; registered automatically by automation_file.core.action_registry.build_default_registry(). Supports plain FTP and explicit FTPS (via FTP_TLS + auth()).

FTP delete operations.

automation_file.remote.ftp.delete_ops.ftp_delete_path(remote_path)[source]

Delete a remote file over FTP / FTPS.

Parameters:

remote_path (str)

Return type:

bool

WebDAV

HTTP-based remote storage client. Uses PROPFIND for directory listings and rejects private/loopback targets unless allow_private_hosts=True.

SMB / CIFS

Built on the high-level smbclient API from smbprotocol. Uses UNC paths (\\\\server\\share\\path) under the hood and defaults to encrypted sessions.

SMB / CIFS client built on smbprotocol’s high-level smbclient API.

Scope mirrors automation_file.remote.webdav.client — existence check, upload, download, delete, directory create, and shallow listing. The underlying session is registered per (server, username) pair and torn down when SMBClient.close() runs. smbprotocol is imported lazily so importing this module never touches the optional dependency.

class automation_file.remote.smb.client.SMBClient(server, share, username=None, password=None, *, port=445, encrypt=True, connection_timeout=30.0)[source]

Bases: object

Minimal SMB client scoped to the operations used by this project.

Parameters:

  • server (str)

  • share (str)

  • username (str | None)

  • password (str | None)

  • port (int)

  • encrypt (bool)

  • connection_timeout (float)

close()[source]
Return type:

None

delete(remote_path)[source]

Remove the remote file at remote_path.

Parameters:

remote_path (str)

Return type:

None

download(remote_path, local_path)[source]

Stream the remote resource at remote_path to local_path.

Parameters:
Return type:

None

exists(remote_path)[source]

Return True if the remote path exists.

Parameters:

remote_path (str)

Return type:

bool

list_dir(remote_path)[source]

Return a shallow listing of remote_path (non-recursive).

Parameters:

remote_path (str)

Return type:

list[SMBEntry]

mkdir(remote_path)[source]

Create the remote directory at remote_path (parents must exist).

Parameters:

remote_path (str)

Return type:

None

rmdir(remote_path)[source]

Remove the empty remote directory at remote_path.

Parameters:

remote_path (str)

Return type:

None

upload(local_path, remote_path)[source]

Copy the contents of local_path to remote_path on the share.

Parameters:
Return type:

None

class automation_file.remote.smb.client.SMBEntry(name, is_dir, size)[source]

Bases: object

A single directory listing entry returned by SMBClient.list_dir().

Parameters:
is_dir: bool
name: str
size: int | None

fsspec bridge

Thin wrapper over fsspec so any filesystem it knows about (memory, local, s3, gcs, abfs, …) can be driven through the action registry. No SSRF guard — treat as a developer helper, not a remote-ingestion surface.

Bridge helpers that expose fsspec backends through the same verbs as our native clients.

fsspec already implements a large catalogue of filesystems — memory, local, HTTP, GCS, ABFS, SSH, and more. Rather than reimplement each one, this module gives callers a tiny surface (upload / download / exists / list_dir / delete / mkdir) over any fsspec URL. The fsspec import is lazy so installing the package is only required when the bridge is actually used.

This is a developer helper, not a user-input surface. Callers are responsible for validating URLs before handing them in — there is no SSRF guard here because fsspec supports dozens of schemes, many of which bypass the http(s) validator entirely (ssh://, s3://, gcs://…).

class automation_file.remote.fsspec_bridge.FsspecEntry(name, is_dir, size)[source]

Bases: object

A single directory listing entry returned by fsspec_list_dir().

Parameters:
is_dir: bool
name: str
size: int | None
automation_file.remote.fsspec_bridge.fsspec_delete(url, *, recursive=False)[source]

Remove url from its fsspec filesystem.

Parameters:
Return type:

None

automation_file.remote.fsspec_bridge.fsspec_download(url, local_path)[source]

Download the fsspec resource at url to local_path.

Parameters:
Return type:

None

automation_file.remote.fsspec_bridge.fsspec_exists(url)[source]

Return True if url exists on its backing fsspec filesystem.

Parameters:

url (str)

Return type:

bool

automation_file.remote.fsspec_bridge.fsspec_list_dir(url)[source]

Return a shallow listing of url as FsspecEntry records.

Parameters:

url (str)

Return type:

list[FsspecEntry]

automation_file.remote.fsspec_bridge.fsspec_mkdir(url, *, create_parents=True)[source]

Create the directory at url (optionally including parents).

Parameters:
Return type:

None

automation_file.remote.fsspec_bridge.fsspec_upload(local_path, url)[source]

Copy local_path onto the fsspec target at url.

Parameters:
Return type:

None

automation_file.remote.fsspec_bridge.get_fs(url_or_protocol, **storage_options)[source]

Return an fsspec.AbstractFileSystem for url_or_protocol.

Pass either a bare protocol ("s3", "memory") or a full URL — fsspec’s url_to_fs will extract the protocol and pass storage_options through to the backend constructor.

Parameters:

  • url_or_protocol (str)

  • storage_options (Any)

Return type:

Any

Cross-backend

Cross-backend copy — stream a file from one storage backend to another.

copy_between(source, target) resolves each URI to (backend, parameters), downloads the source to a private temp file, then uploads from that temp file to the target. Every backend already exposes *_download_file / *_upload_file primitives — this module just picks the right pair and cleans up the intermediate temp file.

Supported URI schemes:

  • local:/absolute/path or a bare filesystem path

  • s3://bucket/key

  • azure://container/blob

  • dropbox:/path

  • sftp:/remote/path

  • ftp:/remote/path

  • http://... / https://... (source only)

Callers must have previously initialised every backend they reference (s3_instance.later_init, etc.) — this helper does not manage sessions.

exception automation_file.remote.cross_backend.CrossBackendException[source]

Bases: FileAutomationException

Raised when a URI is malformed or refers to an unknown backend.

automation_file.remote.cross_backend.copy_between(source, target)[source]

Copy the object at source to target via a local temp file.

Returns True when both the download and the upload reported success.

Parameters:
Return type:

bool