From 143d9d269f86b046c48c98a19ebc6433c8f9a00c Mon Sep 17 00:00:00 2001 From: ashishtransfirm Date: Sun, 25 Oct 2026 10:01:10 +0530 Subject: [PATCH] feat: support overriding HTTP headers and User-Agent (#98) - Add parse_headers and get_default_headers helpers in utils.py - Support -H/--header in CLI download, deploy, and delete commands - Support headers parameter in download(), deploy(), and delete() API functions - Pass headers in SPARQLWrapper queries and requests calls - Support headers in declarative workflow steps - Record sanitized headers in manifest parameters - Support DATABUS_USER_AGENT env var and default databusclient/ User-Agent - Add comprehensive unit tests in tests/test_headers.py and update docs --- databusclient/__init__.py | 11 +- databusclient/api/delete.py | 85 ++++- databusclient/api/deploy.py | 36 +- databusclient/api/download.py | 227 ++++++++++--- databusclient/api/utils.py | 144 +++++++- databusclient/cli.py | 98 +++++- databusclient/workflow/steps.py | 88 +++-- doc/cli-usage.md | 9 +- doc/module-usage.md | 16 + tests/test_headers.py | 570 ++++++++++++++++++++++++++++++++ 10 files changed, 1159 insertions(+), 125 deletions(-) create mode 100644 tests/test_headers.py diff --git a/databusclient/__init__.py b/databusclient/__init__.py index 37ef670..bfc449b 100644 --- a/databusclient/__init__.py +++ b/databusclient/__init__.py @@ -7,9 +7,18 @@ from databusclient import cli from databusclient.api.deploy import create_dataset, create_distribution, deploy +from databusclient.api.download import download +from databusclient.api.delete import delete from databusclient.version import __version__ -__all__ = ["__version__", "create_dataset", "deploy", "create_distribution"] +__all__ = [ + "__version__", + "create_dataset", + "deploy", + "create_distribution", + "download", + "delete", +] def run(): diff --git a/databusclient/api/delete.py b/databusclient/api/delete.py index 199e5a4..3668d98 100644 --- a/databusclient/api/delete.py +++ b/databusclient/api/delete.py @@ -14,6 +14,9 @@ from databusclient.api.utils import ( fetch_databus_jsonld, get_databus_id_parts_from_file_url, + get_default_headers, + set_header, + set_header_if_missing, ) @@ -23,16 +26,18 @@ class DeleteQueue: Allows adding multiple databus URIs to a queue and executing their deletion in batch. """ - def __init__(self, databus_key: str, manifest_context=None): + def __init__(self, databus_key: str, manifest_context=None, headers=None): """Create a DeleteQueue bound to a given Databus API key. Args: databus_key: API key used to authenticate deletion requests. manifest_context: Optional ManifestContext to record deletion outcomes into. Passed through to _delete_list on execute(). + headers: Optional custom HTTP headers to send with deletion requests. """ self.databus_key = databus_key self.manifest_context = manifest_context + self.headers = headers self.queue: set[str] = set() def add_uri(self, databusURI: str): @@ -79,6 +84,7 @@ def execute(self): self.databus_key, force=True, manifest_context=self.manifest_context, + headers=self.headers, ) @@ -122,6 +128,7 @@ def _delete_resource( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + headers=None, ): """Delete a single Databus resource (version, artifact, group). @@ -159,8 +166,10 @@ def _delete_resource( return print(f"[DELETE] {databusURI}") - headers = {"accept": "*/*", "X-API-KEY": databus_key} - response = requests.delete(databusURI, headers=headers, timeout=30) + req_headers = get_default_headers(headers) + set_header_if_missing(req_headers, "accept", "*/*") + set_header(req_headers, "X-API-KEY", databus_key) + response = requests.delete(databusURI, headers=req_headers, timeout=30) if response.status_code in (200, 204): print(f"Successfully deleted: {databusURI}") @@ -179,6 +188,7 @@ def _delete_list( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + headers=None, ): """Delete a list of Databus resources. @@ -191,7 +201,7 @@ def _delete_list( """ for databusURI in databusURIs: _delete_resource( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers ) @@ -202,6 +212,7 @@ def _delete_artifact( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + headers=None, ): """Delete an artifact and all its versions. @@ -215,7 +226,7 @@ def _delete_artifact( force: If True, skip confirmation prompt and proceed with deletion. queue: If queue is provided, add the URI to the queue instead of deleting immediately. """ - artifact_body = fetch_databus_jsonld(databusURI, databus_key) + artifact_body = fetch_databus_jsonld(databusURI, databus_key, headers=headers) json_dict = json.loads(artifact_body) versions = json_dict.get("databus:hasVersion") @@ -235,11 +246,19 @@ def _delete_artifact( else: # Delete all versions _delete_list( - version_uris, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + version_uris, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers ) # Finally, delete the artifact itself - _delete_resource(databusURI, databus_key, dry_run=dry_run, force=force, queue=queue,manifest_context=manifest_context) + _delete_resource( + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + headers=headers, + ) def _delete_group( @@ -249,6 +268,7 @@ def _delete_group( force: bool = False, queue: DeleteQueue = None, manifest_context=None, + headers=None, ): """Delete a group and all its artifacts and versions. @@ -262,7 +282,7 @@ def _delete_group( force: If True, skip confirmation prompt and proceed with deletion. queue: If queue is provided, add the URI to the queue instead of deleting immediately. """ - group_body = fetch_databus_jsonld(databusURI, databus_key) + group_body = fetch_databus_jsonld(databusURI, databus_key, headers=headers) json_dict = json.loads(group_body) artifacts = json_dict.get("databus:hasArtifact", []) @@ -279,14 +299,35 @@ def _delete_group( # Delete all artifacts (which deletes their versions) for artifact_uri in artifact_uris: _delete_artifact( - artifact_uri, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + artifact_uri, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + headers=headers, ) # Finally, delete the group itself - _delete_resource(databusURI, databus_key, dry_run=dry_run, force=force, queue=queue,manifest_context=manifest_context) + _delete_resource( + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + headers=headers, + ) -def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool, manifest_context=None): +def delete( + databusURIs: List[str], + databus_key: str, + dry_run: bool = False, + force: bool = False, + manifest_context=None, + headers=None, +): """Delete a dataset from the databus. Delete a group, artifact, or version identified by the given databus URI. @@ -299,7 +340,7 @@ def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool, force: If True, skip confirmation prompt and proceed with deletion. """ - queue = DeleteQueue(databus_key, manifest_context=manifest_context) + queue = DeleteQueue(databus_key, manifest_context=manifest_context, headers=headers) for databusURI in databusURIs: _host, _account, group, artifact, version, file = ( @@ -309,24 +350,36 @@ def delete(databusURIs: List[str], databus_key: str, dry_run: bool, force: bool, if group == "collections" and artifact is not None: print(f"Deleting collection: {databusURI}") _delete_resource( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers ) elif file is not None: print(f"Deleting file is not supported via API: {databusURI}") elif version is not None: print(f"Deleting version: {databusURI}") _delete_resource( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context, headers=headers ) elif artifact is not None: print(f"Deleting artifact and all its versions: {databusURI}") _delete_artifact( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + headers=headers, ) elif group is not None and group != "collections": print(f"Deleting group and all its artifacts and versions: {databusURI}") _delete_group( - databusURI, databus_key, dry_run=dry_run, force=force, queue=queue, manifest_context=manifest_context + databusURI, + databus_key, + dry_run=dry_run, + force=force, + queue=queue, + manifest_context=manifest_context, + headers=headers, ) else: print(f"Deleting {databusURI} is not supported.") diff --git a/databusclient/api/deploy.py b/databusclient/api/deploy.py index 2c8cd08..6f5ac85 100644 --- a/databusclient/api/deploy.py +++ b/databusclient/api/deploy.py @@ -8,10 +8,16 @@ import hashlib import json from enum import Enum -from typing import Dict, List, Optional, Tuple, Union +from typing import Any, Dict, Iterable, List, Optional, Tuple, Union import requests +from databusclient.api.utils import ( + get_default_headers, + set_header, + set_header_if_missing, +) + _debug = False @@ -176,14 +182,17 @@ def _get_file_stats(distribution_str: str) -> Tuple[Optional[str], Optional[int] return sha256sum, content_length -def _load_file_stats(url: str) -> Tuple[str, int]: +def _load_file_stats( + url: str, headers: Union[Dict[str, Any], Iterable[str], None] = None +) -> Tuple[str, int]: """Download the file at ``url`` and compute its SHA-256 and length. This is used as a fallback when the caller did not supply checksum/size information in the CLI or metadata file. """ - resp = requests.get(url, timeout=30) + req_headers = get_default_headers(headers) + resp = requests.get(url, headers=req_headers, timeout=30) if resp.status_code >= 400: raise requests.exceptions.RequestException(response=resp) @@ -192,7 +201,10 @@ def _load_file_stats(url: str) -> Tuple[str, int]: return sha256sum, content_length -def get_file_info(distribution_str: str) -> Tuple[Dict[str, str], str, str, str, int]: +def get_file_info( + distribution_str: str, + headers: Union[Dict[str, Any], Iterable[str], None] = None, +) -> Tuple[Dict[str, str], str, str, str, int]: """Return parsed file information for a distribution string. Returns a tuple `(cvs, format_extension, compression, sha256sum, size)`. @@ -210,7 +222,7 @@ def get_file_info(distribution_str: str) -> Tuple[Dict[str, str], str, str, str, if sha256sum is None or content_length is None: __url = str(distribution_str).split("|")[0] - sha256sum, content_length = _load_file_stats(__url) + sha256sum, content_length = _load_file_stats(__url, headers=headers) return cvs, format_extension, compression, sha256sum, content_length @@ -320,6 +332,7 @@ def create_dataset( group_title: str = None, group_abstract: str = None, group_description: str = None, + headers: Union[Dict[str, Any], Iterable[str], None] = None, ) -> Dict[str, Union[List[Dict[str, Union[bool, str, int, float, List]]], str]]: """ Creates a Databus Dataset as a python dict from distributions and submitted metadata. WARNING: If file stats (sha256sum, content length) @@ -376,7 +389,7 @@ def create_dataset( compression, sha256sum, content_length, - ) = get_file_info(dst_string) + ) = get_file_info(dst_string, headers=headers) if not cvs and len(distributions) > 1: raise BadArgumentException( @@ -463,6 +476,7 @@ def deploy( verify_parts: bool = False, log_level: DeployLogLevel = DeployLogLevel.debug, debug: bool = False, + headers: Union[Dict[str, Any], Iterable[str], None] = None, ) -> None: """Deploys a dataset to the databus. The endpoint is inferred from the DataID identifier. Parameters @@ -479,7 +493,9 @@ def deploy( controls whether output shold be printed to the console (stdout) """ - headers = {"X-API-KEY": f"{api_key}", "Content-Type": "application/json"} + req_headers = get_default_headers(headers) + set_header(req_headers, "X-API-KEY", f"{api_key}") + set_header_if_missing(req_headers, "Content-Type", "application/json") data = json.dumps(dataid) try: @@ -491,7 +507,7 @@ def deploy( base + f"/api/publish?verify-parts={str(verify_parts).lower()}&log-level={log_level.name}" ) - resp = requests.post(api_uri, data=data, headers=headers, timeout=30) + resp = requests.post(api_uri, data=data, headers=req_headers, timeout=30) if debug or _debug: try: @@ -517,6 +533,7 @@ def deploy_from_metadata( artifact_version_description: str, license_url: str, apikey: str, + headers: Union[Dict[str, Any], Iterable[str], None] = None, ) -> None: """ Deploy a dataset from metadata entries. @@ -547,10 +564,11 @@ def deploy_from_metadata( artifact_version_description=artifact_version_description, license_url=license_url, distributions=distributions, + headers=headers, ) print(f"Deploying dataset version: {version_id}") - deploy(dataset, apikey) + deploy(dataset, apikey, headers=headers) print(f"Successfully deployed to {version_id}") print(f"Deployed {len(metadata)} file(s):") diff --git a/databusclient/api/download.py b/databusclient/api/download.py index e2c0755..af7a865 100644 --- a/databusclient/api/download.py +++ b/databusclient/api/download.py @@ -1,4 +1,5 @@ import json +import inspect import os import bz2 import gzip @@ -7,7 +8,7 @@ import re import shutil import tempfile -from urllib.parse import urlparse +from urllib.parse import urljoin, urlparse import requests from SPARQLWrapper import JSON, SPARQLWrapper @@ -20,6 +21,12 @@ fetch_databus_jsonld, get_databus_id_parts_from_file_url, compute_sha256_and_length, + parse_headers, + get_default_headers, + has_header, + set_header, + set_header_if_missing, + strip_sensitive_headers, ) from databusclient.filehandling.format import ( convert_file, @@ -51,6 +58,23 @@ GRAPH_MODES = {"download-url"} +def _origin(url: str) -> Tuple[str, str]: + """Return lowercase (scheme, netloc) for origin comparison.""" + parsed = urlparse(url) + return (parsed.scheme.lower(), parsed.netloc.lower()) + + +def _headers_for_url( + req_headers: dict, + url: str, + trusted_origins: Optional[set] = None, +) -> dict: + """Drop credential headers unless URL belongs to a trusted origin.""" + if trusted_origins is None or _origin(url) in trusted_origins: + return dict(req_headers) + return strip_sensitive_headers(req_headers) + + def _get_download_directory(url: str, localDir: str | None) -> str: """Return the local Databus-layout directory for a file URL.""" _host, account, group, artifact, version, file = get_databus_id_parts_from_file_url( @@ -266,7 +290,9 @@ def _convert_compression_format( raise RuntimeError(f"Compression conversion failed: {e}") -def get_expected_checksum(databusURI: str, databus_key: str) -> str | None: +def get_expected_checksum( + databusURI: str, databus_key: str, headers: dict | None = None +) -> str | None: expected = None host, account, group, artifact, version, file = get_databus_id_parts_from_file_url( @@ -276,7 +302,9 @@ def get_expected_checksum(databusURI: str, databus_key: str) -> str | None: try: if version is not None: version_uri = f"https://{host}/{account}/{group}/{artifact}/{version}" - json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key) + json_str = fetch_databus_jsonld( + version_uri, databus_key=databus_key, headers=headers + ) checks = _extract_checksums_from_jsonld(json_str) expected = checks.get(databusURI) or checks.get( "https://" + databusURI.removeprefix("http://").removeprefix("https://") @@ -393,7 +421,9 @@ def _extract_checksums_from_jsonld(json_str: str) -> dict: return checksums -def _resolve_checksums_for_urls(file_urls: List[str], databus_key: str | None) -> dict: +def _resolve_checksums_for_urls( + file_urls: List[str], databus_key: str | None, headers: dict | None = None +) -> dict: """ Group file URLs by their Version URI, fetch each Version JSON-LD once, and return a combined url->checksum mapping for the provided URLs. @@ -416,7 +446,9 @@ def _resolve_checksums_for_urls(file_urls: List[str], databus_key: str | None) - checksums: dict = {} for version_uri, urls_in_version in versions_map.items(): try: - json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key) + json_str = fetch_databus_jsonld( + version_uri, databus_key=databus_key, headers=headers + ) extracted_checksums = _extract_checksums_from_jsonld(json_str) for url in urls_in_version: if url in extracted_checksums: @@ -443,6 +475,8 @@ def _download_file( expected_checksum: str | None = None, manifest_context=None, user_agent: str | None = None, + headers: dict | None = None, + trusted_origins: set | None = None, ) -> None: """Download a file from the internet with a progress bar using tqdm. @@ -479,23 +513,33 @@ def _download_file( os.makedirs(dirpath, exist_ok=True) # Create the necessary directories # --- 1. Get redirect URL by requesting HEAD --- - headers = {"User-Agent": DEFAULT_USER_AGENT if user_agent is None else user_agent} + req_headers = get_default_headers(headers, user_agent=user_agent) + allowed_origins = ( + set(trusted_origins) + if trusted_origins is not None + else {_origin(source_url)} + ) # --- 1a. public databus --- - response = requests.head(url, headers=headers, timeout=30, allow_redirects=False) - - # Check for redirect and update URL if necessary - if response.headers.get("Location") and response.status_code in [ - 301, - 302, - 303, - 307, - 308, - ]: - url = response.headers.get("Location") + head_headers = _headers_for_url(req_headers, url, allowed_origins) + response = requests.head( + url, headers=head_headers, timeout=30, allow_redirects=False + ) + + # Check for redirects (allow up to 5 hops) while stripping credentials cross-origin + redirects = 0 + while ( + response.headers.get("Location") + and response.status_code in [301, 302, 303, 307, 308] + and redirects < 5 + ): + url = urljoin(url, response.headers.get("Location")) print("Redirects url: ", url) - # Re-do HEAD request on redirect URL - response = requests.head(url, headers=headers, timeout=30) + head_headers = _headers_for_url(req_headers, url, allowed_origins) + response = requests.head( + url, headers=head_headers, timeout=30, allow_redirects=False + ) + redirects += 1 # Extract hostname from final URL (after redirect) to check if vault token needed. # This is the actual download location that may require authentication. @@ -514,19 +558,23 @@ def _download_file( # Token provided; will handle in GET request below else: # Not a vault host; might need databus API key - if not databus_key: + head_headers = _headers_for_url(req_headers, url, allowed_origins) + if not databus_key and not has_header(head_headers, "x-api-key"): raise DownloadAuthError( "Databus API key not given for protected download" ) - headers["X-API-KEY"] = databus_key - response = requests.head(url, headers=headers, timeout=30) + if databus_key and _origin(url) in allowed_origins: + set_header(req_headers, "X-API-KEY", databus_key) + head_headers = _headers_for_url(req_headers, url, allowed_origins) + response = requests.head(url, headers=head_headers, timeout=30) # --- 2. Try direct GET to redirected URL --- - headers["Accept-Encoding"] = ( - "identity" # disable gzip to get correct content-length - ) + get_headers = _headers_for_url(req_headers, url, allowed_origins) + set_header_if_missing( + get_headers, "Accept-Encoding", "identity" + ) # disable gzip to get correct content-length response = requests.get( - url, headers=headers, stream=True, allow_redirects=True, timeout=30 + url, headers=get_headers, stream=True, allow_redirects=True, timeout=30 ) www = response.headers.get( "WWW-Authenticate", "" @@ -555,11 +603,23 @@ def _download_file( # for known hosts. __get_vault_access__ handles reading the refresh # token and exchanging it; errors are translated to DownloadAuthError # for user-friendly CLI output. - vault_token = __get_vault_access__(url, vault_token_file, auth_url, client_id) - headers["Authorization"] = f"Bearer {vault_token}" + vault_kwargs = {} + try: + params = inspect.signature(__get_vault_access__).parameters + if "headers" in params or any( + p.kind == inspect.Parameter.VAR_KEYWORD + for p in params.values() + ): + vault_kwargs["headers"] = req_headers + except (TypeError, ValueError): + pass + vault_token = __get_vault_access__( + url, vault_token_file, auth_url, client_id, **vault_kwargs + ) + get_headers["Authorization"] = f"Bearer {vault_token}" # Retry with token - response = requests.get(url, headers=headers, stream=True, timeout=30) + response = requests.get(url, headers=get_headers, stream=True, timeout=30) # Map common auth failures to friendly messages if response.status_code == 401: @@ -892,6 +952,8 @@ def _download_files( validate_checksum: bool = False, checksums: dict | None = None, user_agent: str | None = None, + headers: dict | None = None, + trusted_origins: set | None = None, ) -> None: """Download multiple files from the databus. @@ -932,35 +994,44 @@ def _download_files( expected_checksum=expected, manifest_context=manifest_context, user_agent=user_agent, + headers=headers, + trusted_origins=trusted_origins, ) -def _get_sparql_query_of_collection(uri: str, databus_key: str | None = None) -> str: +def _get_sparql_query_of_collection( + uri: str, databus_key: str | None = None, headers: dict | None = None +) -> str: """Get SPARQL query of collection members from databus collection URI. Args: uri: The full databus collection URI. databus_key: Optional Databus API key for authentication on protected resources. + headers: Optional custom HTTP headers to send with the request. Returns: SPARQL query string to get download URLs of all files in the collection. """ - headers = {"Accept": "text/sparql"} + req_headers = get_default_headers(headers) + set_header_if_missing(req_headers, "Accept", "text/sparql") if databus_key is not None: - headers["X-API-KEY"] = databus_key + set_header(req_headers, "X-API-KEY", databus_key) - response = requests.get(uri, headers=headers, timeout=30) + response = requests.get(uri, headers=req_headers, timeout=30) response.raise_for_status() return response.text -def _query_sparql_endpoint(endpoint_url, query, databus_key=None) -> dict: +def _query_sparql_endpoint( + endpoint_url, query, databus_key=None, headers=None +) -> dict: """Query a SPARQL endpoint and return results in JSON format. Args: endpoint_url: The URL of the SPARQL endpoint. query: The SPARQL query string. databus_key: Optional API key for authentication. + headers: Optional custom HTTP headers to send with the SPARQL query. Returns: Dictionary containing the query results. @@ -969,14 +1040,30 @@ def _query_sparql_endpoint(endpoint_url, query, databus_key=None) -> dict: sparql.method = "POST" sparql.setQuery(query) sparql.setReturnFormat(JSON) - if databus_key is not None: - sparql.setCustomHttpHeaders({"X-API-KEY": databus_key}) + + req_headers = get_default_headers(headers) + if hasattr(sparql, "setCustomHttpHeaders"): + custom = dict(req_headers) + if databus_key is not None: + custom["X-API-KEY"] = databus_key + sparql.setCustomHttpHeaders(custom) + if hasattr(sparql, "addCustomHttpHeader"): + for k, v in req_headers.items(): + if k.lower() == "user-agent" and hasattr(sparql, "agent"): + sparql.agent = v + sparql.addCustomHttpHeader(k, v) + if databus_key is not None: + sparql.addCustomHttpHeader("X-API-KEY", databus_key) + elif hasattr(sparql, "agent"): + for k, v in req_headers.items(): + if k.lower() == "user-agent": + sparql.agent = v results = sparql.query().convert() return results def _get_file_download_urls_from_sparql_query( - endpoint_url, query, databus_key=None + endpoint_url, query, databus_key=None, headers=None ) -> List[str]: """Execute a SPARQL query to get databus file download URLs. @@ -984,11 +1071,14 @@ def _get_file_download_urls_from_sparql_query( endpoint_url: The URL of the SPARQL endpoint. query: The SPARQL query string. databus_key: Optional API key for authentication. + headers: Optional custom HTTP headers to send with the SPARQL query. Returns: List of file download URLs. """ - result_dict = _query_sparql_endpoint(endpoint_url, query, databus_key=databus_key) + result_dict = _query_sparql_endpoint( + endpoint_url, query, databus_key=databus_key, headers=headers + ) bindings = result_dict.get("results", {}).get("bindings") if not isinstance(bindings, list): @@ -1012,7 +1102,11 @@ def _get_file_download_urls_from_sparql_query( def __get_vault_access__( - download_url: str, token_file: str, auth_url: str, client_id: str + download_url: str, + token_file: str, + auth_url: str, + client_id: str, + headers: dict | None = None, ) -> str: """ Get Vault access token for a protected databus download. @@ -1027,6 +1121,8 @@ def __get_vault_access__( if len(refresh_token) < 80: print(f"Warning: token from {token_file} is short (<80 chars)") + post_headers = strip_sensitive_headers(get_default_headers(headers)) + # 2. Refresh token -> access token resp = requests.post( auth_url, @@ -1035,6 +1131,7 @@ def __get_vault_access__( "grant_type": "refresh_token", "refresh_token": refresh_token, }, + headers=post_headers, timeout=30, ) resp.raise_for_status() @@ -1059,6 +1156,7 @@ def __get_vault_access__( "subject_token": access_token, "audience": audience, }, + headers=post_headers, timeout=30, ) resp.raise_for_status() @@ -1084,6 +1182,8 @@ def _download_collection( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + headers: dict | None = None, + trusted_origins: set | None = None, ) -> None: """Download all files in a databus collection. @@ -1104,15 +1204,15 @@ def _download_collection( user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. """ - query = _get_sparql_query_of_collection(uri, databus_key=databus_key) + query = _get_sparql_query_of_collection(uri, databus_key=databus_key, headers=headers) file_urls = _get_file_download_urls_from_sparql_query( - endpoint, query, databus_key=databus_key + endpoint, query, databus_key=databus_key, headers=headers ) # If checksum validation requested, attempt to build url->checksum mapping checksums: dict = {} if validate_checksum: - checksums = _resolve_checksums_for_urls(list(file_urls), databus_key) + checksums = _resolve_checksums_for_urls(list(file_urls), databus_key, headers=headers) _download_files( list(file_urls), @@ -1130,6 +1230,8 @@ def _download_collection( validate_checksum=validate_checksum, checksums=checksums if checksums else None, user_agent=user_agent, + headers=headers, + trusted_origins=trusted_origins, ) @@ -1148,6 +1250,8 @@ def _download_version( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + headers: dict | None = None, + trusted_origins: set | None = None, ) -> None: """Download all files in a databus artifact version. @@ -1167,7 +1271,7 @@ def _download_version( user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. """ - json_str = fetch_databus_jsonld(uri, databus_key=databus_key) + json_str = fetch_databus_jsonld(uri, databus_key=databus_key, headers=headers) file_urls = _get_file_download_urls_from_artifact_jsonld(json_str) # build url -> checksum mapping from JSON-LD when available checksums: dict = {} @@ -1192,6 +1296,8 @@ def _download_version( validate_checksum=validate_checksum, checksums=checksums, user_agent=user_agent, + headers=headers, + trusted_origins=trusted_origins, ) @@ -1211,6 +1317,8 @@ def _download_artifact( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + headers: dict | None = None, + trusted_origins: set | None = None, ) -> None: """Download files in a databus artifact. @@ -1231,13 +1339,13 @@ def _download_artifact( user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. """ - json_str = fetch_databus_jsonld(uri, databus_key=databus_key) + json_str = fetch_databus_jsonld(uri, databus_key=databus_key, headers=headers) versions = _get_databus_versions_of_artifact(json_str, all_versions=all_versions) if isinstance(versions, str): versions = [versions] for version_uri in versions: print(f"Downloading version: {version_uri}") - json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key) + json_str = fetch_databus_jsonld(version_uri, databus_key=databus_key, headers=headers) file_urls = _get_file_download_urls_from_artifact_jsonld(json_str) # extract checksums for this version checksums: dict = {} @@ -1262,6 +1370,8 @@ def _download_artifact( validate_checksum=validate_checksum, checksums=checksums, user_agent=user_agent, + headers=headers, + trusted_origins=trusted_origins, ) @@ -1342,6 +1452,8 @@ def _download_group( manifest_context=None, validate_checksum: bool = False, user_agent: str | None = None, + headers: dict | None = None, + trusted_origins: set | None = None, ) -> None: """Download files in a databus group. @@ -1362,7 +1474,7 @@ def _download_group( user_agent: HTTP User-Agent for file HEAD and GET requests. None uses the versioned databus-python-client default. """ - json_str = fetch_databus_jsonld(uri, databus_key=databus_key) + json_str = fetch_databus_jsonld(uri, databus_key=databus_key, headers=headers) artifacts = _get_databus_artifacts_of_group(json_str) for artifact_uri in artifacts: print(f"Download artifact: {artifact_uri}") @@ -1382,6 +1494,8 @@ def _download_group( manifest_context=manifest_context, validate_checksum=validate_checksum, user_agent=user_agent, + headers=headers, + trusted_origins=trusted_origins, ) @@ -1436,6 +1550,7 @@ def download( validate_checksum: bool = False, manifest_context=None, user_agent: str | None = None, + headers=None, ) -> None: """Download datasets from databus. @@ -1460,6 +1575,7 @@ def download( the versioned databus-python-client default. """ _validate_graph_mode(graph_mode) + parsed_headers = parse_headers(headers) if headers else None for databusURI in databusURIs: host, account, group, artifact, version, file = get_databus_id_parts_from_file_url(databusURI) @@ -1472,6 +1588,7 @@ def download( if uri_endpoint is None: uri_endpoint = f"https://{host}/sparql" print(f"SPARQL endpoint {uri_endpoint}") + trusted = {_origin(databusURI), _origin(uri_endpoint)} if group == "collections" and artifact is not None: print(f"Downloading collection: {databusURI}") @@ -1491,12 +1608,14 @@ def download( manifest_context=manifest_context, validate_checksum=validate_checksum, user_agent=user_agent, + headers=parsed_headers, + trusted_origins=trusted, ) elif file is not None: print(f"Downloading file: {databusURI}") # Try to fetch expected checksum from the parent Version metadata expected = ( - get_expected_checksum(databusURI, databus_key) + get_expected_checksum(databusURI, databus_key, headers=parsed_headers) if validate_checksum else None ) @@ -1518,6 +1637,8 @@ def download( validate_checksum=validate_checksum, expected_checksum=expected, user_agent=user_agent, + headers=parsed_headers, + trusted_origins={_origin(databusURI)}, ) elif version is not None: print(f"Downloading version: {databusURI}") @@ -1536,6 +1657,8 @@ def download( manifest_context=manifest_context, validate_checksum=validate_checksum, user_agent=user_agent, + headers=parsed_headers, + trusted_origins={_origin(databusURI)}, ) elif artifact is not None: print( @@ -1557,6 +1680,8 @@ def download( manifest_context=manifest_context, validate_checksum=validate_checksum, user_agent=user_agent, + headers=parsed_headers, + trusted_origins={_origin(databusURI)}, ) elif group is not None and group != "collections": print( @@ -1578,6 +1703,8 @@ def download( manifest_context=manifest_context, validate_checksum=validate_checksum, user_agent=user_agent, + headers=parsed_headers, + trusted_origins={_origin(databusURI)}, ) elif account is not None: print("accountId not supported yet") # TODO @@ -1594,13 +1721,13 @@ def download( if uri_endpoint is None: # endpoint is required for queries (--databus) raise ValueError("No endpoint given for query") res = _get_file_download_urls_from_sparql_query( - uri_endpoint, databusURI, databus_key=databus_key + uri_endpoint, databusURI, databus_key=databus_key, headers=parsed_headers ) # If checksum validation requested, try to build url->checksum mapping checksums: dict = {} if validate_checksum: - checksums = _resolve_checksums_for_urls(res, databus_key) + checksums = _resolve_checksums_for_urls(res, databus_key, headers=parsed_headers) if not checksums: print( "WARNING: Checksum validation enabled but no checksums found for query results." @@ -1622,4 +1749,6 @@ def download( validate_checksum=validate_checksum, checksums=checksums if checksums else None, user_agent=user_agent, + headers=parsed_headers, + trusted_origins={_origin(uri_endpoint)}, ) diff --git a/databusclient/api/utils.py b/databusclient/api/utils.py index df4f49d..1f538e1 100644 --- a/databusclient/api/utils.py +++ b/databusclient/api/utils.py @@ -4,10 +4,138 @@ `download`, `deploy` and `delete` modules. """ -from typing import Optional, Tuple +from typing import Any, Dict, Iterable, Mapping, Optional, Tuple, Union import hashlib +import os import requests +SAFE_MANIFEST_HEADERS = frozenset( + {"user-agent", "accept", "accept-encoding", "accept-language", "content-type"} +) +_SENSITIVE_HEADER_MARKERS = ( + "auth", + "token", + "secret", + "key", + "cookie", + "password", + "credential", +) + + +def has_header(headers: Mapping[str, Any], name: str) -> bool: + """Return True if a header name is present (case-insensitive).""" + target = name.lower() + return any(k.lower() == target for k in headers) + + +def set_header_if_missing(headers: Dict[str, str], name: str, value: str) -> None: + """Set header only if not already present (case-insensitive).""" + if not has_header(headers, name): + headers[name] = value + + +def set_header(headers: Dict[str, str], name: str, value: str) -> None: + """Set header while removing any case-variant duplicate keys.""" + target = name.lower() + for existing in [k for k in headers if k.lower() == target]: + if existing != name: + del headers[existing] + headers[name] = value + + +def is_sensitive_header(name: str) -> bool: + """Return True if a header name may carry authentication credentials.""" + lower = name.strip().lower() + return any(marker in lower for marker in _SENSITIVE_HEADER_MARKERS) + + +def strip_sensitive_headers(headers: Mapping[str, str]) -> Dict[str, str]: + """Return a copy of headers with credential-bearing headers removed.""" + return {k: v for k, v in headers.items() if not is_sensitive_header(k)} + + +def sanitize_headers_for_manifest(headers: Mapping[str, str]) -> Dict[str, str]: + """Keep only explicitly safe non-sensitive header values in manifests.""" + return { + k: (v if k.strip().lower() in SAFE_MANIFEST_HEADERS else "***") + for k, v in headers.items() + } + + +def parse_headers( + headers: Union[Dict[str, Any], Iterable[str], None], +) -> Dict[str, str]: + """Parse HTTP headers from dict, list, tuple, or None into a clean dict of strings. + + Supports: + - Dict of headers: e.g. {"User-Agent": "MyBot/1.0", "Accept": "*/*"} + - Iterable of 'Name: Value' strings: e.g. ["User-Agent: MyBot/1.0"] + + Args: + headers: Dict or iterable of header strings, or None. + + Returns: + Dict[str, str]: Parsed headers dictionary. + """ + if not headers: + return {} + + if isinstance(headers, dict): + return {str(k).strip(): str(v).strip() for k, v in headers.items()} + + parsed: Dict[str, str] = {} + for h in headers: + if isinstance(h, str): + if ":" not in h: + raise ValueError( + f"Invalid header format '{h}'. Expected 'Header-Name: Header-Value'." + ) + k, v = h.split(":", 1) + parsed[k.strip()] = v.strip() + else: + raise ValueError(f"Unexpected header type: {type(h).__name__}") + return parsed + + +def get_default_headers( + headers: Union[Dict[str, Any], Iterable[str], None] = None, + user_agent: Optional[str] = None, +) -> Dict[str, str]: + """Return headers dictionary with default User-Agent and any overrides applied. + + Priority for ``User-Agent`` is: explicit header override, explicit + ``user_agent`` argument, ``DATABUS_USER_AGENT`` environment variable, and + finally ``databus-python-client/ (...)``. + + Args: + headers: User-supplied headers (dict or iterable of strings). + user_agent: Optional User-Agent string override. + + Returns: + Dict[str, str]: Resolved headers dictionary. + """ + custom = parse_headers(headers) + + res: Dict[str, str] = {} + if not has_header(custom, "user-agent"): + if user_agent is not None: + res["User-Agent"] = user_agent + else: + env_ua = os.environ.get("DATABUS_USER_AGENT") + if env_ua: + res["User-Agent"] = env_ua + else: + from databusclient.version import __version__ + + res["User-Agent"] = ( + f"databus-python-client/{__version__} " + "(https://github.com/dbpedia/databus-python-client)" + ) + + res.update(custom) + return res + def get_databus_id_parts_from_file_url( uri: str, @@ -40,21 +168,27 @@ def get_databus_id_parts_from_file_url( return tuple(parts[:6]) # return only the first 6 parts -def fetch_databus_jsonld(uri: str, databus_key: str | None = None) -> str: +def fetch_databus_jsonld( + uri: str, + databus_key: str | None = None, + headers: Union[Dict[str, Any], Iterable[str], None] = None, +) -> str: """Fetch the JSON-LD representation of a Databus resource. Args: uri: Full Databus resource URI. databus_key: Optional API key for protected resources. + headers: Optional custom headers to send with the request. Returns: The response body as a string containing JSON-LD. """ - headers = {"Accept": "application/ld+json"} + req_headers = get_default_headers(headers) + set_header_if_missing(req_headers, "Accept", "application/ld+json") if databus_key is not None: - headers["X-API-KEY"] = databus_key - response = requests.get(uri, headers=headers, timeout=30) + set_header(req_headers, "X-API-KEY", databus_key) + response = requests.get(uri, headers=req_headers, timeout=30) response.raise_for_status() return response.text diff --git a/databusclient/cli.py b/databusclient/cli.py index 4b12e47..4fd7527 100644 --- a/databusclient/cli.py +++ b/databusclient/cli.py @@ -8,7 +8,12 @@ import databusclient.api.deploy as api_deploy from databusclient.api.delete import delete as api_delete -from databusclient.api.download import DownloadAuthError, download as api_download, format_download_connection_error +from databusclient.api.download import ( + DownloadAuthError, + download as api_download, + format_download_connection_error, +) +from databusclient.api.utils import parse_headers, sanitize_headers_for_manifest from databusclient.manifest.context import ManifestContext from databusclient.manifest.writer import ManifestWriter from databusclient.manifest.replay import ManifestReplayError, replay_manifest, load_manifest @@ -67,6 +72,13 @@ def app(): "webdav_url", help="WebDAV URL (e.g., https://cloud.example.com/remote.php/webdav)", ) +@click.option( + "-H", + "--header", + "headers", + multiple=True, + help="Custom HTTP header(s) in 'Name: Value' format (e.g. -H 'User-Agent: MyBot/1.0').", +) @click.option( "--manifest", "manifest_path", @@ -89,6 +101,7 @@ def deploy( path, distributions: List[str], manifest_path, + headers: List[str] = (), ): """ Flexible deploy to Databus command supporting three modes:\n @@ -107,10 +120,15 @@ def deploy( "Invalid combination: when using WebDAV/Nextcloud mode, please provide --webdav-url, --remote, and --path together." ) + try: + parsed_headers = parse_headers(headers) if headers else None + except ValueError as e: + raise click.BadParameter(str(e), param_hint="--header") + manifest_context = None if manifest_path: manifest_context = ManifestContext(command="deploy") - manifest_context.record_params({ + params = { "version_id": version_id, "title": title, "abstract": abstract, @@ -118,7 +136,10 @@ def deploy( "license_url": license_url, "distributions": list(distributions) if distributions else [], "metadata_file": metadata_file, - }) + } + if parsed_headers: + params["headers"] = sanitize_headers_for_manifest(parsed_headers) + manifest_context.record_params(params) def _write_manifest(): if manifest_path and manifest_context is not None: @@ -143,11 +164,12 @@ def _write_manifest(): artifact_version_description=description, license_url=license_url, distributions=distributions, + headers=parsed_headers, ) if manifest_context: manifest_context.replay_params["deploy_mode"] = "classic" manifest_context.replay_params["resolved_distributions"] = dataid["@graph"][-1].get("distribution", []) - api_deploy.deploy(dataid=dataid, api_key=apikey) + api_deploy.deploy(dataid=dataid, api_key=apikey, headers=parsed_headers) if manifest_context: for dist in distributions: url = str(dist).split("|")[0] @@ -170,7 +192,14 @@ def _write_manifest(): manifest_context.replay_params["deploy_mode"] = "metadata" manifest_context.replay_params["resolved_metadata"] = metadata api_deploy.deploy_from_metadata( - metadata, version_id, title, abstract, description, license_url, apikey + metadata, + version_id, + title, + abstract, + description, + license_url, + apikey, + headers=parsed_headers, ) if manifest_context: for entry in metadata: @@ -206,7 +235,14 @@ def _write_manifest(): try: metadata = webdav.upload_to_webdav(distributions, remote, path, webdav_url) api_deploy.deploy_from_metadata( - metadata, version_id, title, abstract, description, license_url, apikey + metadata, + version_id, + title, + abstract, + description, + license_url, + apikey, + headers=parsed_headers, ) if manifest_context: for entry in metadata: @@ -319,6 +355,13 @@ def _write_manifest(): default=None, help="Write a JSON-LD manifest of this operation to PATH (e.g. --manifest manifest.jsonld).", ) +@click.option( + "-H", + "--header", + "headers", + multiple=True, + help="Custom HTTP header(s) in 'Name: Value' format (e.g. -H 'User-Agent: MyBot/1.0').", +) @click.option( "--validate-checksum", is_flag=True, help="Validate checksums of downloaded files" ) @@ -343,6 +386,7 @@ def download( validate_checksum, manifest_path, user_agent, + headers: List[str] = (), ): """ Download datasets from databus, optionally using vault access if vault options are provided. @@ -355,6 +399,11 @@ def download( elif databus_key: auth_method = "databus_key" + try: + parsed_headers = parse_headers(headers) if headers else None + except ValueError as e: + raise click.BadParameter(str(e), param_hint="--header") + manifest_context = None if manifest_path: manifest_context = ManifestContext( @@ -363,7 +412,7 @@ def download( auth_method=auth_method, ) # Record safe replay params — sensitive fields excluded - manifest_context.record_params({ + safe_params = { "databusURIs": list(databusuris), "compression": compression, "convert_format": convert_format, @@ -375,7 +424,10 @@ def download( "user_agent": user_agent, "authurl": authurl, "clientid": clientid, - }) + } + if parsed_headers: + safe_params["headers"] = sanitize_headers_for_manifest(parsed_headers) + manifest_context.record_params(safe_params) try: api_download( localDir=localdir, @@ -394,6 +446,7 @@ def download( validate_checksum=validate_checksum, user_agent=user_agent, manifest_context=manifest_context, + headers=parsed_headers, ) except DownloadAuthError as e: if manifest_context: @@ -434,13 +487,27 @@ def download( @click.option( "--force", is_flag=True, help="Force deletion without confirmation prompt" ) +@click.option( + "-H", + "--header", + "headers", + multiple=True, + help="Custom HTTP header(s) in 'Name: Value' format (e.g. -H 'User-Agent: MyBot/1.0').", +) @click.option( "--manifest", "manifest_path", default=None, help="Write a JSON-LD manifest of this operation to PATH.", ) -def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool, manifest_path: str): +def delete( + databusuris: List[str], + databus_key: str, + dry_run: bool, + force: bool, + manifest_path: str, + headers: List[str] = (), +): """ Delete a dataset from the databus. @@ -448,13 +515,21 @@ def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool, Will recursively delete all data associated with the dataset. """ + try: + parsed_headers = parse_headers(headers) if headers else None + except ValueError as e: + raise click.BadParameter(str(e), param_hint="--header") + manifest_context = None if manifest_path: manifest_context = ManifestContext(command="delete") - manifest_context.record_params({ + params = { "databusURIs": list(databusuris), "dry_run": dry_run, - }) + } + if parsed_headers: + params["headers"] = sanitize_headers_for_manifest(parsed_headers) + manifest_context.record_params(params) try: api_delete( @@ -463,6 +538,7 @@ def delete(databusuris: List[str], databus_key: str, dry_run: bool, force: bool, dry_run=dry_run, force=force, manifest_context=manifest_context, + headers=parsed_headers, ) except Exception as exc: if manifest_context: diff --git a/databusclient/workflow/steps.py b/databusclient/workflow/steps.py index 320e616..82a660d 100644 --- a/databusclient/workflow/steps.py +++ b/databusclient/workflow/steps.py @@ -69,22 +69,25 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None: capture_context = context.manifest_context or ManifestContext(command="download") files_before = len(capture_context.files) - api_download( - localDir=local_dir, - endpoint=resolved.get("databus"), - databusURIs=uris, - token=resolved.get("vault_token"), - databus_key=resolved.get("databus_key"), - all_versions=resolved.get("all_versions", False), - compression=resolved.get("convert_to") or resolved.get("compression"), - convert_format=resolved.get("format"), - graph_name=resolved.get("graph_name"), - base_uri=resolved.get("base_uri"), - graph_mode=resolved.get("graph_mode"), - user_agent=resolved.get("user_agent"), - validate_checksum=resolved.get("validate_checksum", False), - manifest_context=capture_context, - ) + dl_kwargs = { + "localDir": local_dir, + "endpoint": resolved.get("databus"), + "databusURIs": uris, + "token": resolved.get("vault_token"), + "databus_key": resolved.get("databus_key"), + "all_versions": resolved.get("all_versions", False), + "compression": resolved.get("convert_to") or resolved.get("compression"), + "convert_format": resolved.get("format"), + "graph_name": resolved.get("graph_name"), + "base_uri": resolved.get("base_uri"), + "graph_mode": resolved.get("graph_mode"), + "user_agent": resolved.get("user_agent"), + "validate_checksum": resolved.get("validate_checksum", False), + "manifest_context": capture_context, + } + if resolved.get("headers"): + dl_kwargs["headers"] = resolved["headers"] + api_download(**dl_kwargs) new_entries = capture_context.files[files_before:] resolved_urls = [e["url"] for e in new_entries if e.get("status") == "success"] @@ -190,8 +193,12 @@ def _run_classic_mode(self, resolved: Dict[str, Any], name: str) -> list: artifact_version_description=resolved["description"], license_url=resolved["license"], distributions=files, + headers=resolved.get("headers"), ) - api_deploy_call(dataid=dataid, api_key=resolved["api_key"]) + deploy_kwargs = {"dataid": dataid, "api_key": resolved["api_key"]} + if resolved.get("headers"): + deploy_kwargs["headers"] = resolved["headers"] + api_deploy_call(**deploy_kwargs) return files def _run_webdav_mode(self, resolved: Dict[str, Any], name: str) -> list: @@ -207,15 +214,27 @@ def _run_webdav_mode(self, resolved: Dict[str, Any], name: str) -> list: metadata = webdav.upload_to_webdav( local_files, resolved["remote"], resolved["path"], resolved["webdav_url"] ) - deploy_from_metadata( - metadata, - resolved["version_id"], - resolved["title"], - resolved["abstract"], - resolved["description"], - resolved["license"], - resolved["api_key"], - ) + if resolved.get("headers"): + deploy_from_metadata( + metadata, + resolved["version_id"], + resolved["title"], + resolved["abstract"], + resolved["description"], + resolved["license"], + resolved["api_key"], + headers=resolved["headers"], + ) + else: + deploy_from_metadata( + metadata, + resolved["version_id"], + resolved["title"], + resolved["abstract"], + resolved["description"], + resolved["license"], + resolved["api_key"], + ) return [entry.get("url", "") for entry in metadata] @@ -242,13 +261,16 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None: if not api_key: raise StepValidationError(f"Step '{name}': delete step requires 'api_key'.") - api_delete( - databusURIs=uris, - databus_key=api_key, - dry_run=resolved.get("dry_run", False), - force=True, - manifest_context=context.manifest_context, - ) + delete_kwargs = { + "databusURIs": uris, + "databus_key": api_key, + "dry_run": resolved.get("dry_run", False), + "force": True, + "manifest_context": context.manifest_context, + } + if resolved.get("headers"): + delete_kwargs["headers"] = resolved["headers"] + api_delete(**delete_kwargs) context.set_output(name, "output_files", []) diff --git a/doc/cli-usage.md b/doc/cli-usage.md index 2682648..17b8644 100644 --- a/doc/cli-usage.md +++ b/doc/cli-usage.md @@ -1,4 +1,4 @@ -## CLI Usage +## CLI Usage To get started with the command-line interface (CLI) of the databus-python-client, you can use either the Python installation or the Docker image. The examples below show python installation method. @@ -42,6 +42,8 @@ databusclient download $DOWNLOADTARGET - Validates the checksums of downloaded files against the checksums provided by the Databus. If a checksum does not match, an error is raised and the file is deleted. - `--user-agent` - Overrides the HTTP User-Agent header for file download requests. Default: `databus-python-client/ (https://github.com/dbpedia/databus-python-client)`. Example: `--user-agent "DatabusDownloadBot/1.0 (mailto:you@example.org)"`. +- `-H, --header` + - Custom HTTP header(s) to send with requests, in `'Name: Value'` format. Can be specified multiple times (e.g. `-H 'User-Agent: MyBot/1.0 (contact@example.org)'`). Required when accessing servers (such as Wikidata/Wikimedia) that enforce specific User-Agent policies. You can also configure the default User-Agent via the `DATABUS_USER_AGENT` environment variable. **Help and further information on download command:** ```bash @@ -80,6 +82,11 @@ databusclient download https://databus.dbpedia.org/dbpedia/collections/dbpedia-s databusclient download 'PREFIX dcat: SELECT ?x WHERE { ?sub dcat:downloadURL ?x . } LIMIT 10' --databus https://databus.dbpedia.org/sparql ``` +**Download with Custom Headers / User-Agent**: send specific headers required by servers (such as Wikidata) +```bash +databusclient download https://databus.dbpedia.org/dbpedia/dbpedia-wikidata-kg-dump -H "User-Agent: MyBot/1.0 (contact@example.org)" +``` + **Download with Compression Conversion**: download files and convert compression format on-the-fly. Source compression is auto-detected from the file extension. ```bash # Convert all compressed files to gzip format diff --git a/doc/module-usage.md b/doc/module-usage.md index 68a8282..491352a 100644 --- a/doc/module-usage.md +++ b/doc/module-usage.md @@ -80,3 +80,19 @@ deploy(dataset, "mysterious API key") ``` The API key can be found or generated in the Databus account settings. + + +### Download with Custom Headers + +When querying or downloading datasets from servers that require specific HTTP headers (such as Wikidata requiring a custom User-Agent): + +```python +from databusclient import download + +download( + localDir="./downloads", + endpoint=None, + databusURIs=["https://databus.dbpedia.org/dbpedia/dbpedia-wikidata-kg-dump"], + headers={"User-Agent": "MyBot/1.0 (https://example.org/bot; contact@example.org)"}, +) +``` diff --git a/tests/test_headers.py b/tests/test_headers.py new file mode 100644 index 0000000..5886d0f --- /dev/null +++ b/tests/test_headers.py @@ -0,0 +1,570 @@ +"""Tests for overriding HTTP headers and User-Agent configuration.""" + +import pytest +from click.testing import CliRunner + +from databusclient.api.utils import parse_headers, get_default_headers, fetch_databus_jsonld +from databusclient.api.delete import DeleteQueue, delete as api_delete +from databusclient.api.deploy import create_dataset, deploy_from_metadata +from databusclient.api.download import ( + _download_file, + _query_sparql_endpoint, + __get_vault_access__, + download as api_download, +) +from databusclient.cli import app +from databusclient.version import __version__ +from databusclient.workflow.context import StepContext +from databusclient.workflow.steps import DownloadStep + + +def test_parse_headers_none_and_empty(): + assert parse_headers(None) == {} + assert parse_headers([]) == {} + assert parse_headers({}) == {} + + +def test_parse_headers_dict(): + input_dict = {"User-Agent": "MyBot/1.0", "X-Custom": "val123"} + parsed = parse_headers(input_dict) + assert parsed == {"User-Agent": "MyBot/1.0", "X-Custom": "val123"} + # Verify non-string types get stringified + assert parse_headers({"X-Num": 42}) == {"X-Num": "42"} + + +def test_parse_headers_iterable_strings(): + header_list = [ + "User-Agent: MyBot/1.0", + "Authorization: Bearer token123", + "Accept: application/json", + ] + parsed = parse_headers(header_list) + assert parsed == { + "User-Agent": "MyBot/1.0", + "Authorization": "Bearer token123", + "Accept": "application/json", + } + + +def test_parse_headers_with_colons_in_value(): + headers = [ + "User-Agent: MyBot/1.0 (https://example.org/bot; contact: admin@example.org)", + "X-Custom-Time: 2026-10-25T10:00:00Z", + ] + parsed = parse_headers(headers) + assert ( + parsed["User-Agent"] + == "MyBot/1.0 (https://example.org/bot; contact: admin@example.org)" + ) + assert parsed["X-Custom-Time"] == "2026-10-25T10:00:00Z" + + +def test_parse_headers_invalid_format(): + with pytest.raises(ValueError, match="Invalid header format"): + parse_headers(["InvalidHeaderWithoutColon"]) + + with pytest.raises(ValueError, match="Unexpected header type"): + parse_headers([123]) + + +def test_get_default_headers_without_overrides(monkeypatch): + monkeypatch.delenv("DATABUS_USER_AGENT", raising=False) + headers = get_default_headers() + assert headers["User-Agent"] == ( + f"databus-python-client/{__version__} " + "(https://github.com/dbpedia/databus-python-client)" + ) + + +def test_get_default_headers_env_var_override(monkeypatch): + monkeypatch.setenv("DATABUS_USER_AGENT", "EnvBot/2.0") + headers = get_default_headers() + assert headers["User-Agent"] == "EnvBot/2.0" + + +def test_get_default_headers_explicit_override(): + headers = get_default_headers({"User-Agent": "WikidataBot/1.0"}) + assert headers["User-Agent"] == "WikidataBot/1.0" + + +def test_get_default_headers_case_insensitive_user_agent(): + headers = get_default_headers(["user-agent: CustomCaseBot/1.0"]) + # Should not duplicate User-Agent + assert any(k.lower() == "user-agent" for k in headers) + ua_keys = [k for k in headers if k.lower() == "user-agent"] + assert len(ua_keys) == 1 + assert headers[ua_keys[0]] == "CustomCaseBot/1.0" + + +def test_get_default_headers_preserves_custom_headers(): + headers = get_default_headers({"X-App-ID": "test-app", "Accept-Language": "de"}) + assert headers["X-App-ID"] == "test-app" + assert headers["Accept-Language"] == "de" + assert "User-Agent" in headers + + +def test_fetch_databus_jsonld_sends_headers(monkeypatch): + captured_headers = {} + + class FakeResponse: + status_code = 200 + text = '{"@id": "https://databus.dbpedia.org/test"}' + + def raise_for_status(self): + pass + + def fake_get(url, headers=None, timeout=30): + captured_headers.update(headers or {}) + return FakeResponse() + + monkeypatch.setattr("requests.get", fake_get) + fetch_databus_jsonld( + "https://databus.dbpedia.org/test", + headers={"User-Agent": "JsonLdBot/1.0", "X-Trace": "abc"}, + ) + + assert captured_headers.get("User-Agent") == "JsonLdBot/1.0" + assert captured_headers.get("X-Trace") == "abc" + assert captured_headers.get("Accept") == "application/ld+json" + + +def test_download_file_sends_custom_headers(tmp_path, monkeypatch): + head_headers = {} + get_headers = {} + + class FakeHead: + status_code = 200 + headers = {} + + class FakeGet: + status_code = 200 + headers = {"content-length": "13"} + + def iter_content(self, chunk_size): + yield b"dummy content" + + def raise_for_status(self): + pass + + def fake_head(url, headers=None, timeout=30, allow_redirects=False): + head_headers.update(headers or {}) + return FakeHead() + + def fake_get(url, headers=None, stream=True, allow_redirects=True, timeout=30): + get_headers.update(headers or {}) + return FakeGet() + + monkeypatch.setattr("requests.head", fake_head) + monkeypatch.setattr("requests.get", fake_get) + + target_url = "https://databus.dbpedia.org/org/group/art/1.0/file.ttl" + _download_file( + url=target_url, + localDir=str(tmp_path), + headers={"User-Agent": "CustomDownloader/1.0", "X-Custom-Auth": "secret123"}, + ) + + assert head_headers.get("User-Agent") == "CustomDownloader/1.0" + assert head_headers.get("X-Custom-Auth") == "secret123" + assert get_headers.get("User-Agent") == "CustomDownloader/1.0" + assert get_headers.get("X-Custom-Auth") == "secret123" + + +def test_query_sparql_endpoint_sets_headers(monkeypatch): + class MockSPARQL: + def __init__(self, endpoint): + self.endpoint = endpoint + self.method = None + self.customHttpHeaders = {} + self.agent = None + + def setQuery(self, q): + pass + + def setReturnFormat(self, f): + pass + + def addCustomHttpHeader(self, k, v): + self.customHttpHeaders[k] = v + + def query(self): + class R: + def convert(self): + return {"results": {"bindings": []}} + + return R() + + monkeypatch.setattr("databusclient.api.download.SPARQLWrapper", MockSPARQL) + + result = _query_sparql_endpoint( + "https://query.wikidata.org/sparql", + "SELECT * WHERE { ?s ?p ?o } LIMIT 1", + headers={"User-Agent": "WikidataSPARQLBot/1.0", "X-Request-ID": "req-99"}, + ) + assert result == {"results": {"bindings": []}} + + +def test_cli_download_with_header_flag(tmp_path, monkeypatch): + captured_headers = None + + def fake_download(*args, **kwargs): + nonlocal captured_headers + captured_headers = kwargs.get("headers") + + monkeypatch.setattr("databusclient.cli.api_download", fake_download) + + runner = CliRunner() + result = runner.invoke( + app, + [ + "download", + "https://databus.dbpedia.org/test/artifact/1.0/file.ttl", + "--localdir", + str(tmp_path), + "-H", + "User-Agent: CLIBot/1.0", + "-H", + "X-Extra: value", + ], + ) + + assert result.exit_code == 0 + assert captured_headers == { + "User-Agent": "CLIBot/1.0", + "X-Extra": "value", + } + + +def test_cli_download_invalid_header_error(): + runner = CliRunner() + result = runner.invoke( + app, + [ + "download", + "https://databus.dbpedia.org/test/artifact/1.0/file.ttl", + "-H", + "MalformedHeaderWithoutColon", + ], + ) + assert result.exit_code != 0 + assert "Invalid header format" in result.output + + +def test_manifest_records_redacted_headers(tmp_path, monkeypatch): + def fake_download(*args, **kwargs): + pass + + monkeypatch.setattr("databusclient.cli.api_download", fake_download) + + manifest_file = tmp_path / "test-manifest.jsonld" + runner = CliRunner() + result = runner.invoke( + app, + [ + "download", + "https://databus.dbpedia.org/test/artifact/1.0/file.ttl", + "--manifest", + str(manifest_file), + "-H", + "User-Agent: PublicBot/1.0", + "-H", + "Authorization: Bearer sensitive_token", + "-H", + "X-API-KEY: secret_key", + ], + ) + assert result.exit_code == 0 + assert manifest_file.exists() + + content = manifest_file.read_text(encoding="utf-8") + assert "PublicBot/1.0" in content + assert "sensitive_token" not in content + assert "secret_key" not in content + assert "***" in content + + +def test_workflow_step_executes_with_headers(monkeypatch, tmp_path): + received_kwargs = {} + + def fake_download(*args, **kwargs): + received_kwargs.update(kwargs) + + monkeypatch.setattr("databusclient.workflow.steps.api_download", fake_download) + + step = DownloadStep() + ctx = StepContext() + + step.run( + { + "name": "download_data", + "command": "download", + "uri": "https://databus.dbpedia.org/test/artifact/1.0/file.ttl", + "localdir": str(tmp_path), + "headers": { + "User-Agent": "WorkflowBot/1.0", + "X-Workflow-Run": "42", + }, + }, + ctx, + ) + + assert received_kwargs.get("headers") == { + "User-Agent": "WorkflowBot/1.0", + "X-Workflow-Run": "42", + } + + +def test_fetch_databus_jsonld_preserves_lowercase_accept(monkeypatch): + captured = {} + + class FakeResp: + text = "{}" + + def raise_for_status(self): + pass + + def fake_get(url, headers=None, timeout=30): + captured.update(headers or {}) + return FakeResp() + + monkeypatch.setattr("requests.get", fake_get) + fetch_databus_jsonld( + "https://databus.dbpedia.org/test", + headers={"accept": "application/json"}, + ) + assert captured.get("accept") == "application/json" + assert "Accept" not in captured + + +def test_cross_origin_redirect_strips_credential_headers(tmp_path, monkeypatch): + seen = [] + + class RedirectHead: + status_code = 302 + headers = {"Location": "https://mirror.example.net/file.ttl"} + + class OkHead: + status_code = 200 + headers = {} + + class OkGet: + status_code = 200 + headers = {"content-length": "4"} + + def iter_content(self, chunk_size): + yield b"data" + + def raise_for_status(self): + pass + + head_responses = [RedirectHead(), OkHead()] + + def fake_head(url, headers=None, **kwargs): + seen.append(("HEAD", url, dict(headers or {}))) + return head_responses.pop(0) + + def fake_get(url, headers=None, **kwargs): + seen.append(("GET", url, dict(headers or {}))) + return OkGet() + + monkeypatch.setattr("requests.head", fake_head) + monkeypatch.setattr("requests.get", fake_get) + + _download_file( + "https://databus.dbpedia.org/org/group/art/1.0/file.ttl", + localDir=str(tmp_path), + headers={ + "User-Agent": "MirrorSafeBot/1.0", + "Authorization": "Bearer topsecret", + "X-API-KEY": "secret-key", + "Cookie": "session=xyz", + }, + ) + + assert "Authorization" in seen[0][2] + assert "Authorization" not in seen[1][2] + assert "X-API-KEY" not in seen[1][2] + assert "Cookie" not in seen[1][2] + assert seen[1][2]["User-Agent"] == "MirrorSafeBot/1.0" + assert "Authorization" not in seen[2][2] + assert seen[2][2]["User-Agent"] == "MirrorSafeBot/1.0" + + +def test_vault_access_strips_download_credentials(tmp_path, monkeypatch): + token_file = tmp_path / "refresh.tok" + token_file.write_text("x" * 90, encoding="utf-8") + posted_headers = [] + + class TokenResp: + def raise_for_status(self): + pass + + def json(self): + return {"access_token": "minted-token"} + + def fake_post(url, data=None, headers=None, timeout=30): + posted_headers.append(dict(headers or {})) + return TokenResp() + + monkeypatch.setattr("requests.post", fake_post) + tok = __get_vault_access__( + "https://data.dbpedia.io/org/group/art/1.0/file.ttl", + str(token_file), + "https://auth.example.org/token", + "vault-client", + headers={ + "User-Agent": "VaultBot/1.0", + "Authorization": "Bearer download-only", + "X-API-KEY": "databus-secret", + }, + ) + assert tok == "minted-token" + assert len(posted_headers) == 2 + for hdrs in posted_headers: + assert hdrs["User-Agent"] == "VaultBot/1.0" + assert "Authorization" not in hdrs + assert "X-API-KEY" not in hdrs + + +def test_delete_queue_and_group_forward_headers(monkeypatch): + deleted = [] + fetched = [] + + class FakeResp: + status_code = 200 + text = "ok" + + monkeypatch.setattr( + "databusclient.api.delete.fetch_databus_jsonld", + lambda uri, key=None, headers=None: ( + fetched.append((uri, headers)), + '{"databus:hasArtifact": [{"@id": "https://databus.dbpedia.org/acc/grp/art"}], "databus:hasVersion": [{"@id": "https://databus.dbpedia.org/acc/grp/art/1.0"}]}', + )[1], + ) + monkeypatch.setattr( + "requests.delete", + lambda uri, headers=None, timeout=30: (deleted.append((uri, dict(headers or {}))), FakeResp())[1], + ) + + custom = {"User-Agent": "DeleteBot/1.0", "X-Trace": "del-1"} + q = DeleteQueue("k", headers=custom) + q.add_uri("https://databus.dbpedia.org/acc/grp/art/1.0") + q.execute() + assert deleted[-1][1]["X-Trace"] == "del-1" + + deleted.clear() + api_delete( + ["https://databus.dbpedia.org/acc/grp"], + "k", + force=True, + headers=custom, + ) + assert all(hdrs.get("X-Trace") == "del-1" for _, hdrs in fetched) + assert all(hdrs.get("X-Trace") == "del-1" for _, hdrs in deleted) + + +def test_deploy_helpers_forward_headers(monkeypatch): + seen = [] + + class StatResp: + status_code = 200 + content = b"abc" + + class PostResp: + status_code = 200 + text = "ok" + + monkeypatch.setattr( + "requests.get", + lambda url, headers=None, timeout=30: (seen.append(("GET", dict(headers or {}))), StatResp())[1], + ) + monkeypatch.setattr( + "requests.post", + lambda url, data=None, headers=None, timeout=30: (seen.append(("POST", dict(headers or {}))), PostResp())[1], + ) + + custom = {"User-Agent": "DeployBot/1.0", "X-Deploy": "yes"} + create_dataset( + "https://databus.dbpedia.org/acc/grp/art/1.0", + "Title", + "Abstract", + "Description", + "https://creativecommons.org/licenses/by/4.0/", + ["https://example.org/file.ttl|lang=en|ttl|none"], + headers=custom, + ) + assert seen[-1][0] == "GET" and seen[-1][1]["X-Deploy"] == "yes" + + seen.clear() + deploy_from_metadata( + [{"url": "https://example.org/file.ttl", "checksum": "a" * 64, "size": 10, "file_format": "ttl", "compression": "none"}], + "https://databus.dbpedia.org/acc/grp/art/1.0", + "Title", + "Abstract", + "Description", + "https://creativecommons.org/licenses/by/4.0/", + "api-key", + headers=custom, + ) + assert seen[-1][0] == "POST" and seen[-1][1]["X-Deploy"] == "yes" + + +def test_download_group_forwards_headers(tmp_path, monkeypatch): + head_hdrs = [] + + class FakeResp: + status_code = 200 + headers = {"content-length": "1"} + + def iter_content(self, n): + yield b"x" + + def raise_for_status(self): + pass + + monkeypatch.setattr( + "databusclient.api.download.fetch_databus_jsonld", + lambda uri, databus_key=None, headers=None: ( + '{"databus:hasArtifact": [{"@id": "https://databus.dbpedia.org/acc/grp/art"}], ' + '"databus:hasVersion": [{"@id": "https://databus.dbpedia.org/acc/grp/art/1.0"}], ' + '"@graph": [{"@type": "Part", "file": "https://databus.dbpedia.org/acc/grp/art/1.0/f.ttl"}]}' + ), + ) + monkeypatch.setattr( + "requests.head", + lambda url, headers=None, **kw: (head_hdrs.append(dict(headers or {})), FakeResp())[1], + ) + monkeypatch.setattr("requests.get", lambda url, headers=None, **kw: FakeResp()) + + api_download( + str(tmp_path), + None, + ["https://databus.dbpedia.org/acc/grp"], + headers={"X-Group-Header": "g1"}, + ) + assert head_hdrs and head_hdrs[0]["X-Group-Header"] == "g1" + + +def test_manifest_redacts_cookie_and_arbitrary_headers(tmp_path, monkeypatch): + monkeypatch.setattr("databusclient.cli.api_download", lambda *a, **k: None) + manifest_file = tmp_path / "m.jsonld" + result = CliRunner().invoke( + app, + [ + "download", + "https://databus.dbpedia.org/test/artifact/1.0/file.ttl", + "--manifest", + str(manifest_file), + "-H", + "Cookie: session=supersecret", + "-H", + "Proxy-Authorization: Basic dXNlcjpwYXNz", + "-H", + "X-Auth-Token: customsecret", + ], + ) + assert result.exit_code == 0 + content = manifest_file.read_text(encoding="utf-8") + assert "supersecret" not in content + assert "dXNlcjpwYXNz" not in content + assert "customsecret" not in content