Coverage for python/lsst/resources/http.py: 22%
764 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 03:18 -0700
1# This file is part of lsst-resources.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12from __future__ import annotations
14__all__ = ("HttpResourcePath",)
16import contextlib
17import enum
18import functools
19import io
20import json
21import logging
22import math
23import os
24import os.path
25import random
26import re
27import ssl
28import stat
29from collections.abc import Iterator
30from typing import TYPE_CHECKING, Any, BinaryIO, cast
32try:
33 # Prefer 'defusedxml' (not part of standard library) if available, since
34 # 'xml' is vulnerable to XML bombs.
35 import defusedxml.ElementTree as eTree
36except ImportError:
37 import xml.etree.ElementTree as eTree
39try:
40 import fsspec
41 from aiohttp import ClientSession, ClientTimeout, TCPConnector
42 from fsspec.implementations.http import HTTPFileSystem
43 from fsspec.spec import AbstractFileSystem
44except ImportError:
45 fsspec = None
46 AbstractFileSystem = type
47 HTTPFileSystem = type
49from urllib.parse import parse_qs
51import requests
52from astropy import units as u
53from requests.adapters import HTTPAdapter
54from requests.auth import AuthBase
55from urllib3.util.retry import Retry
57from lsst.utils.timer import time_this
59from ._resourceHandles import ResourceHandleProtocol
60from ._resourceHandles._httpResourceHandle import HttpReadResourceHandle, parse_content_range_header
61from ._resourcePath import ResourcePath
62from .utils import _get_num_workers, get_tempdir
64if TYPE_CHECKING:
65 from .utils import TransactionProtocol
67log = logging.getLogger(__name__)
70def _timeout_from_environment(env_var: str, default_value: float) -> float:
71 """Convert and return a timeout from the value of an environment variable
72 or a default value if the environment variable is not initialized. The
73 value of `env_var` must be a valid `float` otherwise this function raises.
75 Parameters
76 ----------
77 env_var : `str`
78 Environment variable to look for.
79 default_value : `float``
80 Value to return if `env_var` is not defined in the environment.
82 Returns
83 -------
84 _timeout_from_environment : `float`
85 Converted value.
86 """
87 try:
88 timeout = float(os.environ.get(env_var, default_value))
89 except ValueError:
90 raise ValueError(
91 f"Expecting valid timeout value in environment variable {env_var} but found "
92 f"{os.environ.get(env_var)}"
93 ) from None
95 if math.isnan(timeout):
96 raise ValueError(f"Unexpected timeout value NaN found in environment variable {env_var}")
98 return timeout
101@functools.lru_cache
102def _calc_tmpdir_buffer_size(tmpdir: str) -> int:
103 """Compute the block size as 256 blocks of typical size
104 (i.e. 4096 bytes) or 10 times the file system block size,
105 whichever is higher.
107 This is a reasonable compromise between
108 using memory for buffering and the number of system calls
109 issued to read from or write to temporary files.
110 """
111 fsstats = os.statvfs(tmpdir)
112 return max(10 * fsstats.f_bsize, 256 * 4096)
115class HttpResourcePathConfig:
116 """Configuration class to encapsulate the configurable items used by class
117 HttpResourcePath.
118 """
120 # Default timeouts for all HTTP requests (seconds).
121 DEFAULT_TIMEOUT_CONNECT: float = 60.0
122 DEFAULT_TIMEOUT_READ: float = 1_500.0
124 # Default lower and upper bounds for the backoff interval (seconds).
125 # A value in this interval is randomly selected as the backoff factor when
126 # requests need to be retried.
127 DEFAULT_BACKOFF_MIN: float = 1.0
128 DEFAULT_BACKOFF_MAX: float = 3.0
130 # Default number of connections to persist with both the front end and
131 # back end servers.
132 DEFAULT_FRONTEND_PERSISTENT_CONNECTIONS: int = 2
133 DEFAULT_BACKEND_PERSISTENT_CONNECTIONS: int = 1
135 # Accepted digest algorithms
136 ACCEPTED_DIGESTS: list[str] = ["adler32", "md5", "sha-256", "sha-512"]
138 def __init__(self) -> None:
139 self._front_end_connections: int | None = None
140 self._back_end_connections: int | None = None
141 self._digest_algorithm: str | None = None
142 self._send_expect_on_put: bool | None = None
143 self._fsspec_is_enabled: bool | None = None
144 self._timeout: tuple[float, float] | None = None
145 self._collect_memory_usage: bool | None = None
146 self._backoff_min: float | None = None
147 self._backoff_max: float | None = None
148 self._ca_bundle: str | None = ""
149 self._client_token: str | None = ""
150 self._client_cert: str | None = ""
151 self._client_key: str | None = ""
152 self._tmpdir_buffersize: tuple[str, int] | None = None
153 self._ssl_context: ssl.SSLContext | None = None
155 @property
156 def front_end_connections(self) -> int:
157 """Number of persistent connections to the front end server."""
158 if self._front_end_connections is not None: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 return self._front_end_connections
161 default_pool_size = max(_get_num_workers(), self.DEFAULT_FRONTEND_PERSISTENT_CONNECTIONS)
163 try:
164 self._front_end_connections = int(
165 os.environ.get("LSST_HTTP_FRONTEND_PERSISTENT_CONNECTIONS", default_pool_size)
166 )
167 except ValueError:
168 self._front_end_connections = default_pool_size
170 return self._front_end_connections
172 @property
173 def back_end_connections(self) -> int:
174 """Number of persistent connections to the back end servers."""
175 if self._back_end_connections is not None: 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true
176 return self._back_end_connections
178 default_pool_size = max(_get_num_workers(), self.DEFAULT_FRONTEND_PERSISTENT_CONNECTIONS)
180 try:
181 self._back_end_connections = int(
182 os.environ.get("LSST_HTTP_BACKEND_PERSISTENT_CONNECTIONS", default_pool_size)
183 )
184 except ValueError:
185 self._back_end_connections = default_pool_size
187 return self._back_end_connections
189 @property
190 def digest_algorithm(self) -> str:
191 """Algorithm to ask the server to use for computing and recording
192 digests of each file contents in PUT requests.
194 Returns
195 -------
196 digest_algorithm: `str`
197 The name of a digest algorithm or the empty string if no algotihm
198 is configured.
199 """
200 if self._digest_algorithm is not None:
201 return self._digest_algorithm
203 digest = os.environ.get("LSST_HTTP_DIGEST", "").lower()
204 if digest not in self.ACCEPTED_DIGESTS:
205 digest = ""
207 self._digest_algorithm = digest
208 return self._digest_algorithm
210 @property
211 def send_expect_on_put(self) -> bool:
212 """Return True if a "Expect: 100-continue" header is to be sent to
213 the server on each PUT request.
215 Some servers (e.g. dCache) uses this information as an indication that
216 the client knows how to handle redirects to the specific server that
217 will actually receive the data for PUT requests.
218 """
219 if self._send_expect_on_put is not None:
220 return self._send_expect_on_put
222 self._send_expect_on_put = "LSST_HTTP_PUT_SEND_EXPECT_HEADER" in os.environ
223 return self._send_expect_on_put
225 @property
226 def fsspec_is_enabled(self) -> bool:
227 """Return True if `fsspec` is enabled for objects of class
228 HttpResourcePath.
230 To determine if `fsspec` is enabled, this method inspects the presence
231 of the environment variable `LSST_HTTP_ENABLE_FSSPEC` (with any value).
232 """
233 if self._fsspec_is_enabled is not None:
234 return self._fsspec_is_enabled
236 self._fsspec_is_enabled = "LSST_HTTP_ENABLE_FSSPEC" in os.environ
237 return self._fsspec_is_enabled
239 @property
240 def timeout(self) -> tuple[float, float]:
241 """Return a tuple with the values of timeouts for connecting to the
242 server and reading its response, respectively. Both values are in
243 seconds.
244 """
245 if self._timeout is not None:
246 return self._timeout
248 self._timeout = (
249 _timeout_from_environment("LSST_HTTP_TIMEOUT_CONNECT", self.DEFAULT_TIMEOUT_CONNECT),
250 _timeout_from_environment("LSST_HTTP_TIMEOUT_READ", self.DEFAULT_TIMEOUT_READ),
251 )
252 return self._timeout
254 @property
255 def collect_memory_usage(self) -> bool:
256 """Return true if we want to collect memory usage when timing
257 operations against the remote server via the `lsst.utils.time_this`
258 context manager.
259 """
260 if self._collect_memory_usage is not None:
261 return self._collect_memory_usage
263 self._collect_memory_usage = "LSST_HTTP_COLLECT_MEMORY_USAGE" in os.environ
264 return self._collect_memory_usage
266 @property
267 def backoff_min(self) -> float:
268 """Lower bound of the interval from which a backoff factor is randomly
269 selected when retrying requests (seconds).
270 """
271 if self._backoff_min is not None:
272 return self._backoff_min
274 self._backoff_min = self.DEFAULT_BACKOFF_MIN
275 try:
276 backoff_min = float(os.environ.get("LSST_HTTP_BACKOFF_MIN", self.DEFAULT_BACKOFF_MIN))
277 if not math.isnan(backoff_min): 277 ↛ 282line 277 didn't jump to line 282 because the condition on line 277 was always true
278 self._backoff_min = backoff_min
279 except ValueError:
280 pass
282 return self._backoff_min
284 @property
285 def backoff_max(self) -> float:
286 """Upper bound of the interval from which a backoff factor is randomly
287 selected when retrying requests (seconds).
288 """
289 if self._backoff_max is not None:
290 return self._backoff_max
292 self._backoff_max = self.DEFAULT_BACKOFF_MAX
293 try:
294 backoff_max = float(os.environ.get("LSST_HTTP_BACKOFF_MAX", self.DEFAULT_BACKOFF_MAX))
295 if not math.isnan(backoff_max): 295 ↛ 300line 295 didn't jump to line 300 because the condition on line 295 was always true
296 self._backoff_max = backoff_max
297 except ValueError:
298 pass
300 return self._backoff_max
302 @property
303 def ca_bundle(self) -> str | None:
304 """Local path to the certificate bundle file or directory where the
305 certifcates of the trusted authorities are located.
307 Return None if this host's system certificate bundle should be
308 used for authenticating remote servers' certificates.
309 """
310 if self._ca_bundle != "":
311 return self._ca_bundle
313 # If a bundle was specified via the environment variable
314 # 'LSST_HTTP_CACERT_BUNDLE' use it.
315 self._ca_bundle = os.getenv("LSST_HTTP_CACERT_BUNDLE")
316 return self._ca_bundle
318 @property
319 def client_token(self) -> str | None:
320 """Value of a bearer token or path to a local file which contains
321 the bearer token to use for authenticating the client when sending
322 requests to the webDAV or HTTP server.
324 Return None if no bearer token is configured in the environment.
325 """
326 if self._client_token != "":
327 return self._client_token
329 # If environment variable LSST_HTTP_AUTH_BEARER_TOKEN is
330 # initialized use its value as the bearer token.
331 self._client_token = os.getenv("LSST_HTTP_AUTH_BEARER_TOKEN")
332 return self._client_token
334 @property
335 def client_cert_key(self) -> tuple[str | None, str | None]:
336 """Paths to a local file where the client certificate and associated
337 private key are located.
339 Return a tuple (client certificate, private key) or (None, None) if no
340 client certificate is configured via environment variables.
341 """
342 if self._client_cert != "" and self._client_key != "":
343 return (self._client_cert, self._client_key)
345 # If the environment variables LSST_HTTP_AUTH_CLIENT_CERT
346 # and LSST_HTTP_AUTH_CLIENT_KEY are initialized use their values.
347 self._client_cert = os.getenv("LSST_HTTP_AUTH_CLIENT_CERT")
348 self._client_key = os.getenv("LSST_HTTP_AUTH_CLIENT_KEY")
349 if self._client_cert and self._client_key:
350 if not _is_protected(self._client_key):
351 raise PermissionError(
352 f"Private key file at {self._client_key} must be protected for access only by its owner"
353 )
354 return (self._client_cert, self._client_key)
356 # If only the certificate was provided raise.
357 if self._client_cert:
358 raise ValueError(
359 "Environment variable LSST_HTTP_AUTH_CLIENT_KEY must be set to client private key file path"
360 )
362 # If only the private key was provided raise.
363 if self._client_key:
364 raise ValueError(
365 "Environment variable LSST_HTTP_AUTH_CLIENT_CERT must be set to client certificate file path"
366 )
368 # If a X.509 user proxy is available, use it as client credentials.
369 self._client_cert = self._client_key = os.getenv("X509_USER_PROXY")
370 return (self._client_cert, self._client_key)
372 @property
373 def tmpdir_buffersize(self) -> tuple[str, int]:
374 """Return the path to a temporary directory and the preferred buffer
375 size to use when reading or writing files in that directory.
376 """
377 if self._tmpdir_buffersize is not None:
378 return self._tmpdir_buffersize
380 tmpdir = get_tempdir()
382 # Compute the block size as 256 blocks of typical size
383 # (i.e. 4096 bytes) or 10 times the file system block size,
384 # whichever is higher. This is a reasonable compromise between
385 # using memory for buffering and the number of system calls
386 # issued to read from or write to temporary files.
387 bufsize = _calc_tmpdir_buffer_size(tmpdir)
388 self._tmpdir_buffersize = (tmpdir, bufsize)
390 return self._tmpdir_buffersize
392 @property
393 def ssl_context(self) -> ssl.SSLContext:
394 """Return an SSL context equiped with the certificates of the trusted
395 authorities.
396 """
397 if self._ssl_context is None:
398 self._ssl_context = ssl.create_default_context()
399 if self.ca_bundle is not None:
400 if os.path.isdir(self.ca_bundle):
401 self._ssl_context.load_verify_locations(capath=self.ca_bundle)
402 elif os.path.isfile(self.ca_bundle):
403 self._ssl_context.load_verify_locations(cafile=self.ca_bundle)
405 return self._ssl_context
408@functools.lru_cache
409def _get_dav_and_server_headers(path: ResourcePath | str) -> tuple[str | None, str | None]:
410 """Retrieve the "DAV" and "Server" headers sent by the remote server as
411 part of the response to a single "OPTIONS" HTTP request.
413 Parameters
414 ----------
415 path : `ResourcePath` or `str`
416 URL to the resource to be checked.
417 Should preferably refer to the root since the status is shared
418 by all paths in that server.
420 Returns
421 -------
422 _get_dav_and_server_headers : `tuple[str|None, str|None]`
423 Values of the "DAV" and "Server" headers found in the response or
424 None if any of those headers was not part of the response.
425 """
426 try:
427 if not isinstance(path, HttpResourcePath):
428 path = HttpResourcePath(path)
430 config = HttpResourcePathConfig()
431 with SessionStore(config=config).get(path) as session:
432 resp = session.options(
433 str(path), stream=False, timeout=config.timeout, headers=path._extra_headers
434 )
436 dav_header = server_header = None
437 if resp.status_code == requests.codes.ok:
438 dav_header = resp.headers.get("DAV") if "DAV" in resp.headers else None
439 server_header = resp.headers.get("Server") if "Server" in resp.headers else None
441 return (dav_header, server_header)
443 except requests.exceptions.SSLError as e:
444 log.warning(
445 "Environment variable LSST_HTTP_CACERT_BUNDLE can be used to "
446 "specify the path to a bundle of certificate authorities you trust "
447 "which are not included in the default set of trusted authorities "
448 "of this system."
449 )
450 raise e
453class BearerTokenAuth(AuthBase):
454 """Attach a bearer token 'Authorization' header to each request.
456 Parameters
457 ----------
458 token : `str`
459 Can be either the path to a local protected file which contains the
460 value of the token or the token itself.
461 """
463 def __init__(self, token: str):
464 self._token = self._path = None
465 self._mtime: float = -1.0
466 if not token:
467 return
469 self._token = token
470 if os.path.isfile(token):
471 self._path = os.path.abspath(token)
472 if not _is_protected(self._path):
473 raise PermissionError(
474 f"Bearer token file at {self._path} must be protected for access only by its owner"
475 )
476 self._refresh()
478 def _refresh(self) -> None:
479 """Read the token file (if any) if its modification time is more recent
480 than the last time we read it.
481 """
482 if not self._path:
483 return
485 if (mtime := os.stat(self._path).st_mtime) > self._mtime:
486 log.debug("Reading bearer token file at %s", self._path)
487 self._mtime = mtime
488 with open(self._path) as f:
489 self._token = f.read().rstrip("\n")
491 def __call__(self, req: requests.PreparedRequest) -> requests.PreparedRequest:
492 # Only add a bearer token to a request when using secure HTTP.
493 if req.url and req.url.lower().startswith("https://") and self._token:
494 self._refresh()
495 req.headers["Authorization"] = f"Bearer {self._token}"
496 return req
499class SessionStore:
500 """Cache a reusable HTTP client session per endpoint.
502 Parameters
503 ----------
504 config : `HttpResourcePathConfig`
505 Configuration items shared by all instances of HttpResourcePath.
506 num_pools : `int`, optional
507 Number of connection pools to keep: there is one pool per remote
508 host.
509 max_persistent_connections : `int`, optional
510 Maximum number of connections per remote host to persist in each
511 connection pool.
512 backoff_min : `float`, optional
513 Minimum value of the interval to compute the exponential
514 backoff factor when retrying requests (seconds).
515 backoff_max : `float`, optional
516 Maximum value of the interval to compute the exponential
517 backoff factor when retrying requests (seconds).
518 """
520 def __init__(
521 self,
522 config: HttpResourcePathConfig,
523 num_pools: int = 10,
524 max_persistent_connections: int = 1,
525 backoff_min: float = 1.0,
526 backoff_max: float = 3.0,
527 ) -> None:
528 # Dictionary to store the session associated to a given URI. The key
529 # of the dictionary is a root URI and the value is the session.
530 self._sessions: dict[str, requests.Session] = {}
532 # Configuration for all instances of HttpResourcePath objects.
533 self._config = config
535 # See documentation of urllib3 PoolManager class:
536 # https://urllib3.readthedocs.io
537 self._num_pools: int = num_pools
539 # See urllib3 Advanced Usage documentation:
540 # https://urllib3.readthedocs.io/en/stable/advanced-usage.html
541 self._max_persistent_connections: int = max_persistent_connections
543 # Minimum and maximum values of the interval to compute the exponential
544 # backoff factor when retrying requests (seconds).
545 self._backoff_min: float = backoff_min
546 self._backoff_max: float = backoff_max if backoff_max > backoff_min else backoff_min + 1.0
548 def clear(self) -> None:
549 """Destroy all previously created sessions and attempt to close
550 underlying idle network connections.
551 """
552 # Close all sessions and empty the store. Idle network connections
553 # should be closed as a consequence. We don't have means through
554 # the API exposed by Requests to actually force closing the
555 # underlying open sockets.
556 for session in self._sessions.values():
557 session.close()
559 self._sessions.clear()
561 def get(self, rpath: ResourcePath) -> requests.Session:
562 """Retrieve a session for accessing the remote resource at rpath.
564 Parameters
565 ----------
566 rpath : `ResourcePath`
567 URL to a resource at the remote server for which a session is to
568 be retrieved.
570 Notes
571 -----
572 Once a session is created for a given endpoint it is cached and
573 returned every time a session is requested for any path under that same
574 endpoint. For instance, a single session will be cached and shared
575 for paths "https://www.example.org/path/to/file" and
576 "https://www.example.org/any/other/path".
578 Note that "https://www.example.org" and "https://www.example.org:12345"
579 will have different sessions since the port number is not identical.
580 """
581 root_uri = str(rpath.root_uri())
582 if root_uri not in self._sessions:
583 # We don't have yet a session for this endpoint: create a new one.
584 self._sessions[root_uri] = self._make_session(rpath)
586 return self._sessions[root_uri]
588 def _make_session(self, rpath: ResourcePath) -> requests.Session:
589 """Make a new session configured from values from the environment."""
590 session = requests.Session()
591 root_uri = str(rpath.root_uri())
592 log.debug("Creating new HTTP session for endpoint %s ...", root_uri)
593 retries = Retry(
594 # Total number of retries to allow. Takes precedence over other
595 # counts.
596 total=6,
597 # How many connection-related errors to retry on.
598 connect=3,
599 # How many times to retry on read errors.
600 read=3,
601 # Backoff factor to apply between attempts after the second try
602 # (seconds). Compute a random jitter to prevent all the clients
603 # to overwhelm the server by sending requests at the same time.
604 backoff_factor=self._backoff_min + (self._backoff_max - self._backoff_min) * random.random(),
605 # How many times to retry on bad status codes.
606 status=5,
607 # Set of uppercased HTTP method verbs that we should retry on.
608 # We only automatically retry idempotent requests.
609 allowed_methods=frozenset(
610 [
611 "COPY",
612 "DELETE",
613 "GET",
614 "HEAD",
615 "MKCOL",
616 "OPTIONS",
617 "PROPFIND",
618 "PUT",
619 ]
620 ),
621 # HTTP status codes that we should force a retry on.
622 status_forcelist=frozenset(
623 [
624 requests.codes.too_many_requests, # 429
625 requests.codes.internal_server_error, # 500
626 requests.codes.bad_gateway, # 502
627 requests.codes.service_unavailable, # 503
628 requests.codes.gateway_timeout, # 504
629 ]
630 ),
631 # Whether to respect Retry-After header on status codes defined
632 # above.
633 respect_retry_after_header=True,
634 )
636 # Persist the specified number of connections to the front end server.
637 session.mount(
638 root_uri,
639 HTTPAdapter(
640 pool_connections=self._num_pools,
641 pool_maxsize=self._max_persistent_connections,
642 pool_block=False,
643 max_retries=retries,
644 ),
645 )
647 # Do not persist the connections to back end servers which may vary
648 # from request to request. Systematically persisting connections to
649 # those servers may exhaust their capabilities when there are thousands
650 # of simultaneous clients.
651 session.mount(
652 f"{rpath.scheme}://",
653 HTTPAdapter(
654 pool_connections=self._num_pools,
655 pool_maxsize=0,
656 pool_block=False,
657 max_retries=retries,
658 ),
659 )
661 # If the remote endpoint doesn't use secure HTTP we don't include
662 # bearer tokens in the requests nor need to authenticate the remote
663 # server.
664 if rpath.scheme != "https":
665 return session
667 # Set the trusted CA certificates bundle for authenticating remote
668 # servers.
669 session.verify = True if self._config.ca_bundle is None else self._config.ca_bundle
671 # Should we use a bearer token for client authentication?
672 if (token := self._config.client_token) is not None:
673 log.debug("... using bearer token authentication")
674 session.auth = BearerTokenAuth(token)
675 return session
677 # Should we instead use client certificate and private key?
678 client_cert, client_key = self._config.client_cert_key
679 if client_cert and client_key:
680 log.debug("... using client certificate authentication.")
681 session.cert = (client_cert, client_key)
682 return session
684 log.debug(
685 "Neither LSST_HTTP_AUTH_BEARER_TOKEN nor (LSST_HTTP_AUTH_CLIENT_CERT and "
686 "LSST_HTTP_AUTH_CLIENT_KEY) are initialized. Client authentication is disabled."
687 )
688 return session
691class ActivityCaveat(enum.Enum):
692 """Helper class for enumerating accepted activity caveats for requesting
693 macaroons.
694 """
696 DOWNLOAD = 1
697 UPLOAD = 2
700class HttpResourcePath(ResourcePath):
701 """General HTTP(S) resource.
703 Notes
704 -----
705 In order to configure the behavior of instances of this class, the
706 environment variables below are inspected:
708 - LSST_HTTP_CACERT_BUNDLE: path to a .pem file or to a directory which
709 contains the .pem files of the trusted certificate authorities's
710 certificates. If the remote server presents a server certificate
711 issued by one of those trusted authorities, we trust it.
712 If this environment variable is not initialized, the default
713 authorities of the the execution host are trusted.
715 - LSST_HTTP_AUTH_BEARER_TOKEN: value of a bearer token or path to a
716 local file containing a bearer token to be used as the client
717 authentication mechanism with all requests.
718 The permissions of the token file must be set so that only its
719 owner can access it.
720 If initialized, takes precedence over LSST_HTTP_AUTH_CLIENT_CERT
721 and LSST_HTTP_AUTH_CLIENT_KEY.
723 - LSST_HTTP_AUTH_CLIENT_CERT: path to a .pem file which contains the
724 client certificate for authenticating to the server.
725 If initialized, the variable LSST_HTTP_AUTH_CLIENT_KEY must also be
726 initialized with the path of the client private key file.
727 The permissions of the client private key must be set so that only
728 its owner can access it, at least for reading.
730 - LSST_HTTP_PUT_SEND_EXPECT_HEADER: if set (with any value), a
731 "Expect: 100-Continue" header will be added to all HTTP PUT requests.
732 This header is required by some servers to detect if the client
733 knows how to handle redirections. In case of redirection, the body
734 of the PUT request is sent to the redirected location and not to
735 the front end server.
737 - LSST_HTTP_TIMEOUT_CONNECT and LSST_HTTP_TIMEOUT_READ: if set to a
738 numeric value, they are interpreted as the number of seconds to wait
739 for establishing a connection with the server and for reading its
740 response, respectively.
742 - LSST_HTTP_FRONTEND_PERSISTENT_CONNECTIONS and
743 LSST_HTTP_BACKEND_PERSISTENT_CONNECTIONS: contain the maximum number
744 of connections to attempt to persist with both the front end servers
745 and the back end servers.
746 Default values: DEFAULT_FRONTEND_PERSISTENT_CONNECTIONS and
747 DEFAULT_BACKEND_PERSISTENT_CONNECTIONS.
749 - LSST_HTTP_DIGEST: case-insensitive name of the digest algorithm to
750 ask the server to compute for every file's content sent to the server
751 via a PUT request. No digest is requested if this variable is not set
752 or is set to an invalid value.
753 Valid values are those in ACCEPTED_DIGESTS.
755 - LSST_HTTP_ENABLE_FSSPEC: the presence of this environment variable
756 activates the usage of `fsspec` compatible file system to read
757 a HTTP URL. The value of the variable is not inspected.
758 """
760 @staticmethod
761 def create_http_resource_path(
762 path: str, *, extra_headers: dict[str, str] | None = None
763 ) -> HttpResourcePath:
764 """Create an instance of `HttpResourcePath` with additional
765 HTTP-specific configuration.
767 Parameters
768 ----------
769 path : `str`
770 HTTP URL to be wrapped in a `ResourcePath` instance.
771 extra_headers : `dict` [ `str`, `str` ], optional
772 Additional headers that will be sent with every HTTP request made
773 by this `ResourcePath`. These override any headers that may be
774 generated internally by `HttpResourcePath` (e.g. authentication
775 headers).
777 Return
778 ------
779 instance : `ResourcePath`
780 Newly-created `HttpResourcePath` instance.
782 Notes
783 -----
784 Most users should use the `ResourcePath` constructor, instead.
785 """
786 # Make sure we instantiate ResourcePath using a string to guarantee we
787 # get a new ResourcePath. If we accidentally provided a ResourcePath
788 # instance instead, the ResourcePath constructor sometimes returns
789 # the original object and we would be modifying an object that is
790 # supposed to be immutable.
791 instance = ResourcePath(str(path))
792 assert isinstance(instance, HttpResourcePath)
793 instance._extra_headers = extra_headers
794 return instance
796 # WebDAV servers known to be able to sign URLs. The values are lowercased
797 # server identifiers retrieved from the 'Server' header included in
798 # the response to a HTTP OPTIONS request.
799 SUPPORTED_URL_SIGNERS = ("dcache", "xrootd")
801 # Configuration items for this class instances.
802 _config: HttpResourcePathConfig = HttpResourcePathConfig()
804 # The session for metadata requests is used for interacting with
805 # the front end servers for requests such as PROPFIND, HEAD, etc. Those
806 # interactions are typically served by the front end servers. We want to
807 # keep the connection to the front end servers open, to reduce the cost
808 # associated to TCP and TLS handshaking for each new request.
809 _metadata_session_store = SessionStore(
810 config=_config,
811 num_pools=5,
812 max_persistent_connections=_config.front_end_connections,
813 backoff_min=_config.backoff_min,
814 backoff_max=_config.backoff_max,
815 )
817 # The data session is used for interaction with the front end servers which
818 # typically redirect to the back end servers for serving our PUT and GET
819 # requests. We attempt to keep a single connection open with the front end
820 # server, if possible. This depends on how the server behaves and the
821 # kind of request. Some servers close the connection when redirecting
822 # the client to a back end server, for instance when serving a PUT
823 # request.
824 _data_session_store = SessionStore(
825 config=_config,
826 num_pools=25,
827 max_persistent_connections=_config.back_end_connections,
828 backoff_min=_config.backoff_min,
829 backoff_max=_config.backoff_max,
830 )
832 # Process ID which created the session stores above. We need to store this
833 # to replace sessions created by a parent process and inherited by a
834 # child process after a fork, to avoid confusing the SSL layer.
835 _pid: int = -1
837 # Connector used by a session pool to establish network connections to
838 # remote servers. This connector is exclusively used by fsspec file system
839 # and is shared by all instances of this class.
840 _tcp_connector: TCPConnector | None = None
842 # Additional headers added to every request.
843 _extra_headers: dict[str, str] | None = None
845 @property
846 def metadata_session(self) -> _SessionWrapper:
847 """Client session to send requests which do not require upload or
848 download of data, i.e. mostly metadata requests.
849 """
850 session = None
851 if hasattr(self, "_metadata_session"):
852 if HttpResourcePath._pid == os.getpid():
853 session = self._metadata_session
854 else:
855 # The metadata session we have in cache was likely created by
856 # a parent process. Discard all the sessions in that store.
857 self._metadata_session_store.clear()
859 # Retrieve a new metadata session.
860 if session is None:
861 HttpResourcePath._pid = os.getpid()
862 session = self._metadata_session_store.get(self)
863 self._metadata_session: requests.Session = session
864 return _SessionWrapper(session, extra_headers=self._extra_headers)
866 @property
867 def data_session(self) -> _SessionWrapper:
868 """Client session for uploading and downloading data."""
869 session = None
870 if hasattr(self, "_data_session"):
871 if HttpResourcePath._pid == os.getpid():
872 session = self._data_session
873 else:
874 # The data session we have in cache was likely created by
875 # a parent process. Discard all the sessions in that store.
876 self._data_session_store.clear()
878 # Retrieve a new data session.
879 if session is None:
880 HttpResourcePath._pid = os.getpid()
881 session = self._data_session_store.get(self)
882 self._data_session: requests.Session = session
883 return _SessionWrapper(session, extra_headers=self._extra_headers)
885 def _clear_sessions(self) -> None:
886 """Close the socket connections that are still open.
888 Used only in test suites to avoid warnings.
889 """
890 self._metadata_session_store.clear()
891 self._data_session_store.clear()
893 if hasattr(self, "_metadata_session"):
894 delattr(self, "_metadata_session")
896 if hasattr(self, "_data_session"):
897 delattr(self, "_data_session")
899 def _init_server_properties(self) -> None:
900 """
901 Initialize instance variables '_is_webdav' and '_server' by
902 sending a single OPTIONS request to the remote server and
903 saving the results.
904 """
905 # Retrieve the "DAV" and the "Server" headers for the root URL of this
906 # path
907 dav_header, server_header = _get_dav_and_server_headers(self.root_uri())
909 # Check that "1" is part of the value of the "DAV" header. We don't
910 # use locks, so a server complying to class 1 is enough for our
911 # purposes. All webDAV servers must advertise at least compliance
912 # class "1".
913 #
914 # Compliance classes are documented in
915 # http://www.webdav.org/specs/rfc4918.html#dav.compliance.classes
916 #
917 # Examples of values for header DAV are:
918 # DAV: 1, 2
919 # DAV: 1, <http://apache.org/dav/propset/fs/1>
920 self._is_webdav: bool = False
921 if dav_header is not None:
922 self._is_webdav = "1" in dav_header.replace(" ", "").split(",")
924 self._server: str | None = None
925 if server_header is not None:
926 # Server header is expected to be of the form 'dCache/9.2.4'
927 # or 'XrootD/v5.7.1'. Strip version and put in lowercase.
928 self._server = server_header.split("/")[0].lower()
930 @property
931 def is_webdav_endpoint(self) -> bool:
932 """Check if the current endpoint implements WebDAV features.
934 This is stored per URI but cached by root so there is only one check
935 per hostname.
936 """
937 if hasattr(self, "_is_webdav"):
938 return self._is_webdav
940 self._init_server_properties()
941 return self._is_webdav
943 @property
944 def server(self) -> str | None:
945 """Return the lowercased identifier of the remote server, retrieved
946 from the response header 'Server' from an 'OPTIONS' HTTP request.
948 If the remote server does not include that header in its response
949 to an 'OPTIONS' request, server() returns None.
951 Examples of return values are "dcache", "xrootd".
952 """
953 if hasattr(self, "_server"):
954 return self._server
956 self._init_server_properties()
957 return self._server
959 @property
960 def server_signs_urls(self) -> bool:
961 """Return true if the remote server support signing or URLs for
962 download and upload.
963 """
964 return self.server in HttpResourcePath.SUPPORTED_URL_SIGNERS
966 @classmethod
967 def _reload_config(cls) -> None:
968 """Reload the configuration for all instances of this class. That
969 configuration is instantiated from the environment.
971 This is an internal method mainly intended for tests.
972 """
973 HttpResourcePath._config = HttpResourcePathConfig()
975 def exists(self) -> bool:
976 """Check that a remote HTTP resource exists."""
977 log.debug("Checking if resource exists: %s", self.geturl())
978 if not self.is_webdav_endpoint:
979 # The remote is a plain HTTP server. Let's attempt a HEAD
980 # request, even if the behavior for such a request against a
981 # directory is not specified, so it depends on the server
982 # implementation.
983 resp = self._head_non_webdav_url()
984 return self._is_successful_non_webdav_head_request(resp)
986 # The remote endpoint is a webDAV server: send a PROPFIND request
987 # to determine if it exists.
988 resp = self._propfind()
989 if resp.status_code == requests.codes.multi_status: # 207
990 prop = _parse_propfind_response_body(resp.text)[0]
991 return prop.exists
992 else: # 404 Not Found
993 return False
995 def size(self) -> int:
996 """Return the size of the remote resource in bytes."""
997 if self.dirLike:
998 return 0
1000 if not self.is_webdav_endpoint:
1001 # The remote is a plain HTTP server. Send a HEAD request to
1002 # retrieve the size of the resource.
1003 resp = self._head_non_webdav_url()
1004 if resp.status_code == requests.codes.ok: # 200
1005 if "Content-Length" in resp.headers:
1006 return int(resp.headers["Content-Length"])
1007 else:
1008 raise ValueError(
1009 f"Response to HEAD request to {self} does not contain 'Content-Length' header"
1010 )
1011 elif resp.status_code == requests.codes.partial_content:
1012 # 206 Partial Content, returned from a GET request with a Range
1013 # header (used to emulate HEAD for presigned S3 URLs).
1014 # In this case Content-Length is the length of the Range and
1015 # not the full length of the file, so we have to parse
1016 # Content-Range instead.
1017 content_range_header = resp.headers.get("Content-Range")
1018 if content_range_header is None:
1019 raise ValueError(
1020 f"Response to GET request to {self} did not contain 'Content-Range' header"
1021 )
1022 content_range = parse_content_range_header(content_range_header)
1023 size = content_range.total
1024 if size is None:
1025 raise ValueError(f"Content-Range header for {self} did not include a total file size")
1026 return size
1027 elif resp.status_code == requests.codes.range_not_satisfiable:
1028 # 416 Range Not Satisfiable, which can occur on a GET for a 0
1029 # byte file since we asked for 1 byte Range which is longer
1030 # than the file.
1031 #
1032 # Servers are supposed to include a Content-Range header in
1033 # this case, but Google's S3 implementation doesn't. Any
1034 # non-zero file size should have been handled by the 206 and
1035 # 200 cases above, so assume we have a zero here.
1036 return 0
1037 elif resp.status_code == requests.codes.not_found:
1038 raise FileNotFoundError(
1039 f"Resource {self} does not exist, status: {resp.status_code} {resp.reason}"
1040 )
1041 else:
1042 raise ValueError(
1043 f"Unexpected response for HEAD request for {self}, status: {resp.status_code} "
1044 f"{resp.reason}"
1045 )
1047 # The remote is a webDAV server: send a PROPFIND request to retrieve
1048 # the size of the resource. Sizes are only meaningful for files.
1049 resp = self._propfind()
1050 if resp.status_code == requests.codes.multi_status: # 207
1051 prop = _parse_propfind_response_body(resp.text)[0]
1052 if prop.is_file:
1053 return prop.size
1054 elif prop.is_directory:
1055 raise IsADirectoryError(
1056 f"Resource {self} is reported by server as a directory but has a file path"
1057 )
1058 else:
1059 raise FileNotFoundError(f"Resource {self} does not exist")
1060 else: # 404 Not Found
1061 raise FileNotFoundError(
1062 f"Resource {self} does not exist, status: {resp.status_code} {resp.reason}"
1063 )
1065 def _head_non_webdav_url(self) -> requests.Response:
1066 """Return a response from a HTTP HEAD request for a non-WebDAV HTTP
1067 URL.
1069 Emulates HEAD using a 1-byte GET for presigned S3 URLs.
1070 """
1071 if self._looks_like_presigned_s3_url():
1072 # Presigned S3 URLs are signed for a single method only, so you
1073 # can't call HEAD on a URL signed for GET. However, S3 does
1074 # support Range requests, so you can ask for a 1-byte range with
1075 # GET for a similar effect to HEAD.
1076 #
1077 # Note that some headers differ between a true HEAD request and the
1078 # response returned by this GET, e.g. Content-Length will always be
1079 # 1, and the status code is 206 instead of 200.
1080 return self.metadata_session.get(
1081 self.geturl(),
1082 timeout=self._config.timeout,
1083 allow_redirects=True,
1084 stream=False,
1085 headers={"Range": "bytes=0-0"},
1086 )
1087 else:
1088 return self.metadata_session.head(
1089 self.geturl(), timeout=self._config.timeout, allow_redirects=True, stream=False
1090 )
1092 def _is_successful_non_webdav_head_request(self, resp: requests.Response) -> bool:
1093 """Return `True` if the status code in the response indicates a
1094 successful response to ``_head_non_webdav_url``.
1095 """
1096 return resp.status_code in (
1097 requests.codes.ok, # 200, from a normal HEAD or GET request
1098 requests.codes.partial_content, # 206, returned from a GET request with a Range header.
1099 # 416, returned from a GET request with a 1-byte Range header that
1100 # is longer than the 0-byte file.
1101 requests.codes.range_not_satisfiable,
1102 )
1104 def _looks_like_presigned_s3_url(self) -> bool:
1105 """Return `True` if this ResourcePath's URL is likely to be a presigned
1106 S3 URL.
1107 """
1108 query_params = parse_qs(self._uri.query)
1109 return "Signature" in query_params and "Expires" in query_params
1111 def mkdir(self) -> None:
1112 """Create the directory resource if it does not already exist."""
1113 # Creating directories is only available on WebDAV back ends.
1114 if not self.is_webdav_endpoint:
1115 raise NotImplementedError(
1116 f"Creation of directory {self} is not implemented by plain HTTP servers"
1117 )
1119 if not self.dirLike:
1120 raise NotADirectoryError(f"Can not create a 'directory' for file-like URI {self}")
1122 # Check if the target directory already exists.
1123 resp = self._propfind()
1124 if resp.status_code == requests.codes.multi_status: # 207
1125 prop = _parse_propfind_response_body(resp.text)[0]
1126 if prop.exists:
1127 if prop.is_directory:
1128 return
1129 else:
1130 # A file exists at this path
1131 raise NotADirectoryError(
1132 f"Can not create a directory for {self} because a file already exists at that path"
1133 )
1135 # Target directory does not exist. Create it and its ancestors as
1136 # needed. We need to test if parent URL is different from self URL,
1137 # otherwise we could be stuck in a recursive loop
1138 # where self == parent.
1139 if self.geturl() != self.parent().geturl():
1140 self.parent().mkdir()
1142 log.debug("Creating new directory: %s", self.geturl())
1143 self._mkcol()
1145 def remove(self) -> None:
1146 """Remove the resource."""
1147 self._delete()
1149 def read(self, size: int = -1) -> bytes:
1150 """Open the resource and return the contents in bytes.
1152 Parameters
1153 ----------
1154 size : `int`, optional
1155 The number of bytes to read. Negative or omitted indicates
1156 that all data should be read.
1157 """
1158 # Use the data session as a context manager to ensure that the
1159 # network connections to both the front end and back end servers are
1160 # closed after downloading the data.
1161 log.debug("Reading from remote resource: %s", self.geturl())
1162 stream = size > 0
1163 with self.data_session as session:
1164 with time_this(log, msg="GET %s", args=(self,)):
1165 resp = session.get(self.geturl(), stream=stream, timeout=self._config.timeout)
1167 if resp.status_code != requests.codes.ok: # 200
1168 raise FileNotFoundError(
1169 f"Unable to read resource {self}; status: {resp.status_code} {resp.reason}"
1170 )
1171 if not stream:
1172 return resp.content
1173 else:
1174 return next(resp.iter_content(chunk_size=size))
1176 def write(self, data: bytes, overwrite: bool = True) -> None:
1177 """Write the supplied bytes to the new resource.
1179 Parameters
1180 ----------
1181 data : `bytes`
1182 The bytes to write to the resource. The entire contents of the
1183 resource will be replaced.
1184 overwrite : `bool`, optional
1185 If `True` the resource will be overwritten if it exists. Otherwise
1186 the write will fail.
1187 """
1188 log.debug("Writing to remote resource: %s", self.geturl())
1189 if not overwrite and self.exists():
1190 raise FileExistsError(f"Remote resource {self} exists and overwrite has been disabled")
1192 # Ensure the parent directory exists.
1193 # This is only meaningful and appropriate for WebDAV, not the general
1194 # HTTP case. e.g. for S3 HTTP URLs, the underlying service has no
1195 # concept of 'directories' at all.
1196 if self.is_webdav_endpoint:
1197 self.parent().mkdir()
1199 # Upload the data.
1200 log.debug("Writing data to remote resource: %s", self.geturl())
1201 self._put(data=data)
1203 def transfer_from(
1204 self,
1205 src: ResourcePath,
1206 transfer: str = "copy",
1207 overwrite: bool = False,
1208 transaction: TransactionProtocol | None = None,
1209 multithreaded: bool = True,
1210 ) -> None:
1211 """Transfer the current resource to a Webdav repository.
1213 Parameters
1214 ----------
1215 src : `ResourcePath`
1216 Source URI.
1217 transfer : `str`
1218 Mode to use for transferring the resource. Supports the following
1219 options: copy.
1220 overwrite : `bool`, optional
1221 Whether overwriting the remote resource is allowed or not.
1222 transaction : `~lsst.resources.utils.TransactionProtocol`, optional
1223 Currently unused.
1224 multithreaded : `bool`, optional
1225 If `True` the transfer will be allowed to attempt to improve
1226 throughput by using parallel download streams. This may of no
1227 effect if the URI scheme does not support parallel streams or
1228 if a global override has been applied. If `False` parallel
1229 streams will be disabled.
1230 """
1231 # Fail early to prevent delays if remote resources are requested.
1232 if transfer not in self.transferModes:
1233 raise ValueError(f"Transfer mode {transfer} not supported by URI scheme {self.scheme}")
1235 # Existence checks cost time so do not call this unless we know
1236 # that debugging is enabled.
1237 if log.isEnabledFor(logging.DEBUG):
1238 log.debug(
1239 "Transferring %s [exists: %s] -> %s [exists: %s] (transfer=%s)",
1240 src,
1241 src.exists(),
1242 self,
1243 self.exists(),
1244 transfer,
1245 )
1247 # Short circuit immediately if the URIs are identical.
1248 if self == src:
1249 log.debug(
1250 "Target and destination URIs are identical: %s, returning immediately."
1251 " No further action required.",
1252 self,
1253 )
1254 return
1256 if not overwrite and self.exists():
1257 raise FileExistsError(f"Destination path {self} already exists.")
1259 if transfer == "auto":
1260 transfer = self.transferDefault
1262 # We can use webDAV 'COPY' or 'MOVE' if both the current and source
1263 # resources are located in the same server.
1264 if isinstance(src, type(self)) and self.root_uri() == src.root_uri() and self.is_webdav_endpoint:
1265 log.debug("Transfer from %s to %s directly", src, self)
1266 return self._move(src) if transfer == "move" else self._copy(src)
1268 # For resources of different classes or for plain HTTP resources we can
1269 # perform the copy or move operation by downloading to a local file
1270 # and uploading to the destination.
1271 self._copy_via_local(src)
1273 # This was an explicit move, try to remove the source.
1274 if transfer == "move":
1275 src.remove()
1277 def walk(
1278 self, file_filter: str | re.Pattern | None = None
1279 ) -> Iterator[list | tuple[ResourcePath, list[str], list[str]]]:
1280 """Walk the directory tree returning matching files and directories.
1282 Parameters
1283 ----------
1284 file_filter : `str` or `re.Pattern`, optional
1285 Regex to filter out files from the list before it is returned.
1287 Yields
1288 ------
1289 dirpath : `ResourcePath`
1290 Current directory being examined.
1291 dirnames : `list` of `str`
1292 Names of subdirectories within dirpath.
1293 filenames : `list` of `str`
1294 Names of all the files within dirpath.
1295 """
1296 if not self.dirLike:
1297 raise ValueError("Can not walk a non-directory URI")
1299 # Walking directories is only available on WebDAV back ends.
1300 if not self.is_webdav_endpoint:
1301 raise NotImplementedError(f"Walking directory {self} is not implemented by plain HTTP servers")
1303 if isinstance(file_filter, str):
1304 file_filter = re.compile(file_filter)
1306 resp = self._propfind(depth="1")
1307 if resp.status_code == requests.codes.multi_status: # 207
1308 files: list[str] = []
1309 dirs: list[str] = []
1311 for prop in _parse_propfind_response_body(resp.text):
1312 if prop.is_file:
1313 files.append(prop.name)
1314 elif not prop.href.rstrip("/").endswith(self.path.rstrip("/")):
1315 # Only include the names of sub-directories not the name of
1316 # the directory being walked.
1317 dirs.append(prop.name)
1319 if file_filter is not None:
1320 files = [f for f in files if file_filter.search(f)]
1322 if not dirs and not files:
1323 return
1324 else:
1325 yield type(self)(self, forceAbsolute=False, forceDirectory=True), dirs, files
1327 for dir in dirs:
1328 new_uri = self.join(dir, forceDirectory=True)
1329 yield from new_uri.walk(file_filter)
1331 def generate_presigned_get_url(self, *, expiration_time_seconds: int) -> str:
1332 """Return a pre-signed URL that can be used to retrieve this resource
1333 using an HTTP GET without supplying any access credentials.
1335 Parameters
1336 ----------
1337 expiration_time_seconds : `int`
1338 Number of seconds until the generated URL is no longer valid.
1340 Returns
1341 -------
1342 url : `str`
1343 HTTP URL signed for GET.
1344 """
1345 if not self.is_webdav_endpoint:
1346 # This is already an HTTP URL readable without any authentication
1347 # credentials, so return it as-is.
1348 return str(self)
1350 return self._sign_with_macaroon(ActivityCaveat.DOWNLOAD, expiration_time_seconds)
1352 def generate_presigned_put_url(self, *, expiration_time_seconds: int) -> str:
1353 """Return a pre-signed URL that can be used to upload a file to this
1354 path using an HTTP PUT without supplying any access credentials.
1356 Parameters
1357 ----------
1358 expiration_time_seconds : `int`
1359 Number of seconds until the generated URL is no longer valid.
1361 Returns
1362 -------
1363 url : `str`
1364 HTTP URL signed for PUT.
1365 """
1366 if not self.is_webdav_endpoint:
1367 return super().generate_presigned_put_url(expiration_time_seconds=expiration_time_seconds)
1369 return self._sign_with_macaroon(ActivityCaveat.UPLOAD, expiration_time_seconds)
1371 def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
1372 """Return an abstract file system and path that can be used by fsspec.
1374 Returns
1375 -------
1376 fs : `fsspec.spec.AbstractFileSystem`
1377 A file system object suitable for use with the returned path.
1378 path : `str`
1379 A path that can be opened by the file system object.
1380 """
1381 if fsspec is None:
1382 return super().to_fsspec()
1384 if not self.is_webdav_endpoint or self.server not in HttpResourcePath.SUPPORTED_URL_SIGNERS:
1385 return fsspec.url_to_fs(self.geturl(), client_kwargs={"headers": self._extra_headers})
1387 if self.isdir():
1388 raise NotImplementedError(
1389 f"method HttpResourcePath.to_fsspec() not implemented for directory {self}"
1390 )
1392 # If usage of fsspec-compatible file system is disabled in the
1393 # configuration we raise an exception which signals the caller
1394 # that it cannot use fsspec. An example of such a caller is
1395 # `lsst.daf.butler.formatters.ParquetFormatter`.
1396 #
1397 # Note that we don't call super().to_fsspec() since that method
1398 # assumes that fsspec can be used provided fsspec package is
1399 # importable.
1400 #
1401 # The motivation for making this configurable is that for HTTP
1402 # URLs fsspec.HTTPFileSystem uses async I/O and we have found
1403 # unexpected behavior by clients when used against dCache for reading
1404 # parquet files via a ParquetFormatter instance. That behavior cannot
1405 # be reproduced when using other callers.
1406 #
1407 # This needs more investigation to discard the possibility that async
1408 # I/O, used by fsspec.HTTPFileSystem, is related to this behavior.
1409 if not self._config.fsspec_is_enabled:
1410 raise ImportError("fsspec is disabled for HttpResourcePath objects with webDAV back end")
1412 async def get_client_session(**kwargs: Any) -> ClientSession:
1413 """Return a aiohttp.ClientSession configured to use an
1414 `aiohttp.TCPConnector` shared by all instances of this class.
1416 Parameters
1417 ----------
1418 **kwargs : `Any`
1419 Keyword arguments passed unmodified to the contructor of
1420 `aiohttp.ClientSession`.
1422 Returns
1423 -------
1424 session : `aiohttp.ClientSession`
1425 Client session that `aiohttp.HTTPFileSystem` will use to pool
1426 TCP connections to the server.
1427 """
1428 if HttpResourcePath._tcp_connector is None:
1429 HttpResourcePath._tcp_connector = TCPConnector(
1430 # SSL context equipped with client credentials and
1431 # configured to validate server certificates.
1432 ssl=self._config.ssl_context,
1433 # Total number of simultaneous connections this connector
1434 # keeps open with any host.
1435 #
1436 # The default is 100 but we deliberately reduced it to
1437 # avoid keeping a large number of open connexions to file
1438 # servers when thousands of quanta execute simultaneously.
1439 #
1440 # In any case, new connexions are automatically established
1441 # when needed.
1442 limit=10,
1443 # Number of simultaneous connections to a single host:port.
1444 limit_per_host=1,
1445 # Close network connection after usage
1446 force_close=True,
1447 )
1449 connect_timeout, read_timeout = self._config.timeout
1450 return ClientSession(
1451 connector=HttpResourcePath._tcp_connector,
1452 timeout=ClientTimeout(
1453 connect=connect_timeout,
1454 sock_connect=connect_timeout,
1455 sock_read=read_timeout,
1456 total=2 * read_timeout,
1457 ),
1458 **kwargs,
1459 )
1461 # Retrieve a signed URL for download valid for 2 hours.
1462 url = self.generate_presigned_get_url(expiration_time_seconds=2 * 3_600)
1464 # HTTPFileSystem constructor accepts the argument 'block_size'. The
1465 # default value is 'fsspec.utils.DEFAULT_BLOCK_SIZE' which is 5 MB.
1466 # That seems to be a reasonable block size for downloading files.
1467 return HTTPFileSystem(get_client=get_client_session), url
1469 def _sign_with_macaroon(self, activity: ActivityCaveat, expiration_time_seconds: int) -> str:
1470 # dCache and XRootD webDAV servers support delivery of macaroons.
1471 #
1472 # For details about dCache macaroons see:
1473 # https://www.dcache.org/manuals/UserGuide-9.2/macaroons.shtml
1474 if self.server is None:
1475 raise NotImplementedError(f"server for '{self}' does not support signing URLs")
1476 elif self.server not in HttpResourcePath.SUPPORTED_URL_SIGNERS:
1477 raise NotImplementedError(f"server '{self.server}' does not support signing for {self}")
1479 match activity:
1480 case ActivityCaveat.DOWNLOAD:
1481 activity_caveat = "DOWNLOAD,LIST"
1482 case ActivityCaveat.UPLOAD:
1483 activity_caveat = "UPLOAD,LIST"
1485 # Retrieve a macaroon for the requested activities and duration
1486 headers = {"Content-Type": "application/macaroon-request"}
1487 body = {
1488 "caveats": [
1489 f"activity:{activity_caveat}",
1490 ],
1491 "validity": f"PT{expiration_time_seconds}S",
1492 }
1493 resp = self._post(data=json.dumps(body), headers=headers)
1494 if resp.status_code != requests.codes.ok:
1495 raise ValueError(
1496 f"could not retrieve a macaroon for URL {self}, status: {resp.status_code} {resp.reason}"
1497 )
1499 # We are expecting the body of the response to be formatted in JSON.
1500 # dCache sets the 'Content-Type' of the response to 'application/json'
1501 # but XRootD does not set any 'Content-Type' header 8-[
1502 #
1503 # An example of a response body returned by dCache is shown below:
1504 # {
1505 # "macaroon": "MDA[...]Qo",
1506 # "uri": {
1507 # "targetWithMacaroon": "https://dcache.example.org/?authz=MD...",
1508 # "baseWithMacaroon": "https://dcache.example.org/?authz=MD...",
1509 # "target": "https://dcache.example.org/",
1510 # "base": "https://dcache.example.org/"
1511 # }
1512 # }
1513 #
1514 # An example of a response body returned by XRootD is shown below:
1515 # {
1516 # "macaroon": "MDA[...]Qo",
1517 # "expires_in": 86400
1518 # }
1519 try:
1520 response_body = json.loads(resp.text)
1521 if "macaroon" in response_body:
1522 return str(self.replace(query=f"authz={response_body['macaroon']}"))
1523 else:
1524 raise ValueError(f"could not retrieve macaroon for URL {self}")
1525 except json.JSONDecodeError:
1526 raise ValueError(f"could not deserialize response to POST request for URL {self}")
1528 @contextlib.contextmanager
1529 def _as_local(
1530 self, multithreaded: bool = True, tmpdir: ResourcePath | None = None
1531 ) -> Iterator[ResourcePath]:
1532 """Download object over HTTP and place in temporary directory.
1534 Parameters
1535 ----------
1536 multithreaded : `bool`, optional
1537 If `True` the transfer will be allowed to attempt to improve
1538 throughput by using parallel download streams. This may of no
1539 effect if the URI scheme does not support parallel streams or
1540 if a global override has been applied. If `False` parallel
1541 streams will be disabled.
1542 tmpdir : `ResourcePath` or `None`, optional
1543 Explicit override of the temporary directory to use for remote
1544 downloads.
1546 Returns
1547 -------
1548 local_uri : `ResourcePath`
1549 A URI to a local POSIX file corresponding to a local temporary
1550 downloaded copy of the resource.
1551 """
1552 # Use the session as a context manager to ensure that connections
1553 # to both the front end and back end servers are closed after the
1554 # download operation is finished.
1555 with self.data_session as session:
1556 resp = session.get(self.geturl(), stream=True, timeout=self._config.timeout)
1557 if resp.status_code != requests.codes.ok:
1558 raise FileNotFoundError(
1559 f"Unable to download resource {self}; status: {resp.status_code} {resp.reason}"
1560 )
1562 if tmpdir is None:
1563 temp_dir, buffer_size = self._config.tmpdir_buffersize
1564 tmpdir = ResourcePath(temp_dir, forceDirectory=True)
1565 else:
1566 buffer_size = _calc_tmpdir_buffer_size(tmpdir.ospath)
1568 with ResourcePath.temporary_uri(
1569 suffix=self.getExtension(), prefix=tmpdir, delete=True
1570 ) as tmp_uri:
1571 expected_length = int(resp.headers.get("Content-Length", "-1"))
1572 with time_this(
1573 log,
1574 msg="GET %s [length=%d] to local file %s [chunk_size=%d]",
1575 args=(self, expected_length, tmp_uri, buffer_size),
1576 mem_usage=self._config.collect_memory_usage,
1577 mem_unit=u.mebibyte,
1578 ):
1579 content_length = 0
1580 with open(tmp_uri.ospath, "wb", buffering=buffer_size) as tmpFile:
1581 for chunk in resp.iter_content(chunk_size=buffer_size):
1582 tmpFile.write(chunk)
1583 content_length += len(chunk)
1585 # Check that the expected and actual content lengths match.
1586 # Perform this check only when the contents of the file was not
1587 # encoded by the server.
1588 if (
1589 "Content-Encoding" not in resp.headers
1590 and expected_length >= 0
1591 and expected_length != content_length
1592 ):
1593 raise ValueError(
1594 f"Size of downloaded file does not match value in Content-Length header for {self}: "
1595 f"expecting {expected_length} and got {content_length} bytes"
1596 )
1598 yield tmp_uri
1600 def _send_webdav_request(
1601 self,
1602 method: str,
1603 url: str | None = None,
1604 headers: dict[str, str] | None = None,
1605 body: str | None = None,
1606 session: _SessionWrapper | None = None,
1607 timeout: tuple[float, float] | None = None,
1608 ) -> requests.Response:
1609 """Send a webDAV request and correctly handle redirects.
1611 Parameters
1612 ----------
1613 method : `str`
1614 The mthod of the HTTP request to be sent, e.g. PROPFIND, MKCOL.
1615 headers : `dict`, optional
1616 A dictionary of key-value pairs (both strings) to include as
1617 headers in the request.
1618 body : `str`, optional
1619 The body of the request.
1621 Notes
1622 -----
1623 This way of sending webDAV requests is necessary for handling
1624 redirection ourselves, since the 'requests' package changes the method
1625 of the redirected request when the server responds with status 302 and
1626 the method of the original request is not HEAD (which is the case for
1627 webDAV requests).
1629 That means that when the webDAV server we interact with responds with
1630 a redirection to a PROPFIND or MKCOL request, the request gets
1631 converted to a GET request when sent to the redirected location.
1633 See `requests.sessions.SessionRedirectMixin.rebuild_method()` in
1634 https://github.com/psf/requests/blob/main/requests/sessions.py
1636 This behavior of the 'requests' package is meant to be compatible with
1637 what is specified in RFC 9110:
1639 https://www.rfc-editor.org/rfc/rfc9110#name-302-found
1641 For our purposes, we do need to follow the redirection and send a new
1642 request using the same HTTP verb.
1643 """
1644 if url is None:
1645 url = self.geturl()
1647 if headers is None:
1648 headers = {}
1650 if session is None:
1651 session = self.metadata_session
1653 if timeout is None:
1654 timeout = self._config.timeout
1656 with time_this(
1657 log,
1658 msg="%s %s",
1659 args=(
1660 method,
1661 url,
1662 ),
1663 mem_usage=self._config.collect_memory_usage,
1664 mem_unit=u.mebibyte,
1665 ):
1666 for _ in range(max_redirects := 5):
1667 resp = session.request(
1668 method,
1669 url,
1670 data=body,
1671 headers=headers,
1672 stream=False,
1673 timeout=timeout,
1674 allow_redirects=False,
1675 )
1676 if resp.is_redirect:
1677 url = resp.headers["Location"]
1678 else:
1679 return resp
1681 # We reached the maximum allowed number of redirects.
1682 # Stop trying.
1683 raise ValueError(
1684 f"Could not get a response to {method} request for {self} after {max_redirects} redirections"
1685 )
1687 def _propfind(self, body: str | None = None, depth: str = "0") -> requests.Response:
1688 """Send a PROPFIND webDAV request and return the response.
1690 Parameters
1691 ----------
1692 body : `str`, optional
1693 The body of the PROPFIND request to send to the server. If
1694 provided, it is expected to be a XML document.
1695 depth : `str`, optional
1696 The value of the 'Depth' header to include in the request.
1698 Returns
1699 -------
1700 response : `requests.Response`
1701 Response to the PROPFIND request.
1703 Notes
1704 -----
1705 It raises `ValueError` if the status code of the PROPFIND request
1706 is different from "207 Multistatus" or "404 Not Found".
1707 """
1708 if body is None:
1709 # Request only the DAV live properties we are explicitly interested
1710 # in namely 'resourcetype', 'getcontentlength', 'getlastmodified'
1711 # and 'displayname'.
1712 body = (
1713 """<?xml version="1.0" encoding="utf-8" ?>"""
1714 """<D:propfind xmlns:D="DAV:"><D:prop>"""
1715 """<D:resourcetype/><D:getcontentlength/><D:getlastmodified/><D:displayname/>"""
1716 """</D:prop></D:propfind>"""
1717 )
1718 headers = {
1719 "Depth": depth,
1720 "Content-Type": 'application/xml; charset="utf-8"',
1721 "Content-Length": str(len(body)),
1722 }
1723 resp = self._send_webdav_request("PROPFIND", headers=headers, body=body)
1724 if resp.status_code in (requests.codes.multi_status, requests.codes.not_found):
1725 return resp
1726 else:
1727 raise ValueError(
1728 f"Unexpected response for PROPFIND request for {self}, status: {resp.status_code} "
1729 f"{resp.reason}"
1730 )
1732 def _options(self) -> requests.Response:
1733 """Send a OPTIONS webDAV request for this resource."""
1734 resp = self._send_webdav_request("OPTIONS")
1735 if resp.status_code in (requests.codes.ok, requests.codes.created):
1736 return resp
1738 raise ValueError(
1739 f"Unexpected response to OPTIONS request for {self}, status: {resp.status_code} {resp.reason}"
1740 )
1742 def _head(self) -> requests.Response:
1743 """Send a HEAD request for this resource."""
1744 if not self.is_webdav_endpoint:
1745 # The remote is a plain HTTP server.
1746 return self._head_non_webdav_url()
1747 return self._send_webdav_request("HEAD")
1749 def _mkcol(self) -> None:
1750 """Send a MKCOL webDAV request to create a collection. The collection
1751 may already exist.
1752 """
1753 resp = self._send_webdav_request("MKCOL")
1754 if resp.status_code == requests.codes.created: # 201
1755 return
1757 if resp.status_code == requests.codes.method_not_allowed: # 405
1758 # The remote directory already exists
1759 log.debug("Can not create directory: %s may already exist: skipping.", self.geturl())
1760 else:
1761 raise ValueError(f"Can not create directory {self}, status: {resp.status_code} {resp.reason}")
1763 def _delete(self) -> None:
1764 """Send a DELETE webDAV request for this resource."""
1765 log.debug("Deleting %s ...", self.geturl())
1767 # If this is a directory, ensure the remote is a webDAV server because
1768 # plain HTTP servers don't support DELETE requests on non-file
1769 # paths.
1770 if self.dirLike and not self.is_webdav_endpoint:
1771 raise NotImplementedError(
1772 f"Deletion of directory {self} is not implemented by plain HTTP servers"
1773 )
1775 # Deleting non-empty directories may take some time, so increase
1776 # the timeout for getting a response from the server.
1777 timeout = self._config.timeout
1778 if self.dirLike:
1779 timeout = (timeout[0], timeout[1] * 100)
1780 resp = self._send_webdav_request("DELETE", timeout=timeout)
1781 if resp.status_code in (
1782 requests.codes.ok,
1783 requests.codes.accepted,
1784 requests.codes.no_content,
1785 requests.codes.not_found,
1786 ):
1787 # We can get a "404 Not Found" error when the file or directory
1788 # does not exist or when the DELETE request was retried several
1789 # times and a previous attempt actually deleted the resource.
1790 # Therefore we consider that a "Not Found" response is not an
1791 # error since we reached the state desired by the user.
1792 return
1793 else:
1794 # TODO: the response to a DELETE request against a webDAV server
1795 # may be multistatus. If so, we need to parse the reponse body to
1796 # determine more precisely the reason of the failure (e.g. a lock)
1797 # and provide a more helpful error message.
1798 raise ValueError(f"Unable to delete resource {self}; status: {resp.status_code} {resp.reason}")
1800 def _copy_via_local(self, src: ResourcePath) -> None:
1801 """Replace the contents of this resource with the contents of a remote
1802 resource by using a local temporary file.
1804 Parameters
1805 ----------
1806 src : `HttpResourcePath`
1807 The source of the contents to copy to `self`.
1808 """
1809 with src.as_local() as local_uri:
1810 log.debug("Transfer from %s to %s via local file %s", src, self, local_uri)
1811 with open(local_uri.ospath, "rb") as f:
1812 self._put(data=f)
1814 def _copy_or_move(self, method: str, src: HttpResourcePath) -> None:
1815 """Send a COPY or MOVE webDAV request to copy or replace the contents
1816 of this resource with the contents of another resource located in the
1817 same server.
1819 Parameters
1820 ----------
1821 method : `str`
1822 The method to perform. Valid values are "COPY" or "MOVE" (in
1823 uppercase).
1824 src : `HttpResourcePath`
1825 The source of the contents to move to `self`.
1826 """
1827 headers = {"Destination": self.geturl()}
1828 resp = self._send_webdav_request(method, url=src.geturl(), headers=headers, session=self.data_session)
1829 if resp.status_code in (requests.codes.created, requests.codes.no_content):
1830 return
1832 if resp.status_code == requests.codes.multi_status:
1833 tree = eTree.fromstring(resp.content)
1834 status_element = tree.find("./{DAV:}response/{DAV:}status")
1835 status = status_element.text if status_element is not None else "unknown"
1836 error = tree.find("./{DAV:}response/{DAV:}error")
1837 raise ValueError(f"{method} returned multistatus reponse with status {status} and error {error}")
1838 else:
1839 raise ValueError(
1840 f"{method} operation from {src} to {self} failed, status: {resp.status_code} {resp.reason}"
1841 )
1843 def _copy(self, src: HttpResourcePath) -> None:
1844 """Send a COPY webDAV request to replace the contents of this resource
1845 (if any) with the contents of another resource located in the same
1846 server.
1848 Parameters
1849 ----------
1850 src : `HttpResourcePath`
1851 The source of the contents to copy to `self`.
1852 """
1853 # Neither dCache nor XrootD currently implement the COPY
1854 # webDAV method as documented in
1855 # http://www.webdav.org/specs/rfc4918.html#METHOD_COPY
1856 # (See issues DM-37603 and DM-37651 for details)
1857 #
1858 # For the time being, we use a temporary local file to
1859 # perform the copy client side.
1860 # TODO: when those 2 issues above are solved remove the 3 lines below.
1861 must_use_local = True
1862 if must_use_local:
1863 return self._copy_via_local(src)
1865 return self._copy_or_move("COPY", src)
1867 def _move(self, src: HttpResourcePath) -> None:
1868 """Send a MOVE webDAV request to replace the contents of this resource
1869 with the contents of another resource located in the same server.
1871 Parameters
1872 ----------
1873 src : `HttpResourcePath`
1874 The source of the contents to move to `self`.
1875 """
1876 return self._copy_or_move("MOVE", src)
1878 def _post(self, data: str | None = None, headers: dict[str, str] | None = None) -> requests.Response:
1879 """Perform an HTTP POST request and returns the received response.
1881 Parameters
1882 ----------
1883 body : `bytes`
1884 The contents of the request body.
1885 """
1886 resp = self.metadata_session.request(
1887 "POST",
1888 self.geturl(),
1889 data=data,
1890 headers=headers,
1891 stream=False,
1892 timeout=self._config.timeout,
1893 allow_redirects=True,
1894 )
1895 if resp.status_code == requests.codes.ok:
1896 return resp
1898 raise ValueError(f"POST request for {self} failed, status: {resp.status_code} {resp.reason}")
1900 def _put(self, data: BinaryIO | bytes) -> None:
1901 """Perform an HTTP PUT request and handle redirection.
1903 Parameters
1904 ----------
1905 data : `Union[BinaryIO, bytes]`
1906 The data to be included in the body of the PUT request.
1907 """
1908 # Retrieve the final URL for this upload by sending a PUT request with
1909 # no content. Follow a single server redirection to retrieve the
1910 # final URL.
1911 headers = {"Content-Length": "0"}
1913 # If we are explicitly configured for or if we know the remote server
1914 # is dCache, send an "Expect" header to signal the server that this
1915 # client knows how to handle redirection in PUT requests.
1916 #
1917 # The goal is that the contents of the file we want to upload is sent
1918 # directly to the dCache pool without transiting through the dCache
1919 # webDAV door. Otherwise, the uploaded data would transit twice over
1920 # the network: first from this client to the dCache webDAV door and
1921 # second from webDAV door to the target pool (i.e. the dCache file
1922 # server which will ultimately store the data we will upload).
1923 #
1924 # Systematically uploading data via dCache webDAV door could add
1925 # unnecessary load to the door which we can avoid by instead uploading
1926 # directly to the dCache pool.
1927 #
1928 # For further details see section "Redirection on upload":
1929 #
1930 # https://www.dcache.org/manuals/UserGuide-9.2/webdav.shtml#redirection
1931 if self._config.send_expect_on_put or self.server == "dcache":
1932 headers["Expect"] = "100-continue"
1934 url = self.geturl()
1936 # Use the session as a context manager to ensure the underlying
1937 # connections are closed after finishing uploading the data.
1938 with self.data_session as session:
1939 # Send an empty PUT request to get redirected to the final
1940 # destination.
1941 log.debug("Sending empty PUT request to %s", url)
1942 with time_this(
1943 log,
1944 msg="PUT (no data) %s",
1945 args=(url,),
1946 mem_usage=self._config.collect_memory_usage,
1947 mem_unit=u.mebibyte,
1948 ):
1949 resp = session.request(
1950 "PUT",
1951 url,
1952 data=None,
1953 headers=headers,
1954 stream=False,
1955 timeout=self._config.timeout,
1956 allow_redirects=False,
1957 )
1958 if resp.is_redirect:
1959 url = resp.headers["Location"]
1961 # Upload the data to the final destination.
1962 log.debug("Uploading data to %s", url)
1964 # Ask the server to compute and record a checksum of the uploaded
1965 # file contents, for later integrity checks. Since we don't compute
1966 # the digest ourselves while uploading the data, we cannot control
1967 # after the request is complete that the data we uploaded is
1968 # identical to the data recorded by the server, but at least the
1969 # server has recorded a digest of the data it stored.
1970 #
1971 # See RFC-3230 for details and
1972 # https://www.iana.org/assignments/http-dig-alg/http-dig-alg.xhtml
1973 # for the list of supported digest algorithhms.
1974 # In addition, note that not all servers implement this RFC so
1975 # the checksum may not be computed by the server.
1976 put_headers: dict[str, str] | None = None
1977 if digest := self._config.digest_algorithm:
1978 put_headers = {"Want-Digest": digest}
1980 with time_this(
1981 log,
1982 msg="PUT %s",
1983 args=(url,),
1984 mem_usage=self._config.collect_memory_usage,
1985 mem_unit=u.mebibyte,
1986 ):
1987 resp = session.request(
1988 "PUT",
1989 url,
1990 data=data,
1991 headers=put_headers,
1992 stream=False,
1993 timeout=self._config.timeout,
1994 allow_redirects=False,
1995 )
1996 if resp.status_code in (
1997 requests.codes.ok,
1998 requests.codes.created,
1999 requests.codes.no_content,
2000 ):
2001 return
2002 else:
2003 raise ValueError(f"Can not write file {self}, status: {resp.status_code} {resp.reason}")
2005 @contextlib.contextmanager
2006 def _openImpl(
2007 self,
2008 mode: str = "r",
2009 *,
2010 encoding: str | None = None,
2011 ) -> Iterator[ResourceHandleProtocol]:
2012 resp = self._head()
2013 accepts_range = resp.status_code == requests.codes.ok and resp.headers.get("Accept-Ranges") == "bytes"
2014 handle: ResourceHandleProtocol
2015 if mode in ("rb", "r") and accepts_range:
2016 handle = HttpReadResourceHandle(mode, log, self, timeout=self._config.timeout)
2017 if mode == "r":
2018 # cast because the protocol is compatible, but does not have
2019 # BytesIO in the inheritance tree
2020 yield io.TextIOWrapper(cast(Any, handle), encoding=encoding)
2021 else:
2022 yield handle
2023 else:
2024 with super()._openImpl(mode, encoding=encoding) as http_handle:
2025 yield http_handle
2027 def _copy_extra_attributes(self, original_uri: ResourcePath) -> None:
2028 assert isinstance(original_uri, HttpResourcePath)
2029 self._extra_headers = original_uri._extra_headers
2032def _dump_response(resp: requests.Response) -> None:
2033 """Log the contents of a HTTP or webDAV request and its response.
2035 Parameters
2036 ----------
2037 resp : `requests.Response`
2038 The response to log.
2040 Notes
2041 -----
2042 Intended for development purposes only.
2043 """
2044 log.debug("-----------------------------------------------")
2045 log.debug("Request")
2046 log.debug(" method=%s", resp.request.method)
2047 log.debug(" URL=%s", resp.request.url)
2048 log.debug(" headers=%s", resp.request.headers)
2049 if resp.request.method == "PUT":
2050 log.debug(" body=<data>")
2051 elif resp.request.body is None:
2052 log.debug(" body=<empty>")
2053 else:
2054 log.debug(" body=%r", resp.request.body[:120])
2056 log.debug("Response:")
2057 log.debug(" status_code=%d", resp.status_code)
2058 log.debug(" headers=%s", resp.headers)
2059 if not resp.content:
2060 log.debug(" body=<empty>")
2061 elif "Content-Type" in resp.headers and resp.headers["Content-Type"] == "text/plain":
2062 log.debug(" body=%r", resp.content)
2063 else:
2064 log.debug(" body=%r", resp.content[:80])
2067def _is_protected(filepath: str) -> bool:
2068 """Return true if the permissions of file at filepath only allow for access
2069 by its owner.
2071 Parameters
2072 ----------
2073 filepath : `str`
2074 Path of a local file.
2075 """
2076 if not os.path.isfile(filepath):
2077 return False
2078 mode = stat.S_IMODE(os.stat(filepath).st_mode)
2079 owner_accessible = bool(mode & stat.S_IRWXU)
2080 group_accessible = bool(mode & stat.S_IRWXG)
2081 other_accessible = bool(mode & stat.S_IRWXO)
2082 return owner_accessible and not group_accessible and not other_accessible
2085def _parse_propfind_response_body(body: str) -> list[DavProperty]:
2086 """Parse the XML-encoded contents of the response body to a webDAV PROPFIND
2087 request.
2089 Parameters
2090 ----------
2091 body : `str`
2092 XML-encoded response body to a PROPFIND request
2094 Returns
2095 -------
2096 responses : `List[DavProperty]`
2098 Notes
2099 -----
2100 Is is expected that there is at least one reponse in `body`, otherwise
2101 this function raises.
2102 """
2103 # A response body to a PROPFIND request is of the form (indented for
2104 # readability):
2105 #
2106 # <?xml version="1.0" encoding="UTF-8"?>
2107 # <D:multistatus xmlns:D="DAV:">
2108 # <D:response>
2109 # <D:href>path/to/resource</D:href>
2110 # <D:propstat>
2111 # <D:prop>
2112 # <D:resourcetype>
2113 # <D:collection xmlns:D="DAV:"/>
2114 # </D:resourcetype>
2115 # <D:getlastmodified>
2116 # Fri, 27 Jan 2 023 13:59:01 GMT
2117 # </D:getlastmodified>
2118 # <D:getcontentlength>
2119 # 12345
2120 # </D:getcontentlength>
2121 # </D:prop>
2122 # <D:status>
2123 # HTTP/1.1 200 OK
2124 # </D:status>
2125 # </D:propstat>
2126 # </D:response>
2127 # <D:response>
2128 # ...
2129 # </D:response>
2130 # <D:response>
2131 # ...
2132 # </D:response>
2133 # </D:multistatus>
2135 # Scan all the 'response' elements and extract the relevant properties
2136 responses = []
2137 multistatus = eTree.fromstring(body.strip())
2138 for response in multistatus.findall("./{DAV:}response"):
2139 responses.append(DavProperty(response))
2141 if responses:
2142 return responses
2143 else:
2144 # Could not parse the body
2145 raise ValueError(f"Unable to parse response for PROPFIND request: {body}")
2148class DavProperty:
2149 """Helper class to encapsulate select live DAV properties of a single
2150 resource, as retrieved via a PROPFIND request.
2152 Parameters
2153 ----------
2154 response : `eTree.Element` or `None`
2155 The XML response defining the DAV property.
2156 """
2158 # Regular expression to compare against the 'status' element of a
2159 # PROPFIND response's 'propstat' element.
2160 _status_ok_rex = re.compile(r"^HTTP/.* 200 .*$", re.IGNORECASE)
2162 def __init__(self, response: eTree.Element | None):
2163 self._href: str = ""
2164 self._displayname: str = ""
2165 self._collection: bool = False
2166 self._getlastmodified: str = ""
2167 self._getcontentlength: int = -1
2169 if response is not None:
2170 self._parse(response)
2172 def _parse(self, response: eTree.Element) -> None:
2173 # Extract 'href'.
2174 if (element := response.find("./{DAV:}href")) is not None:
2175 # We need to use "str(element.text)"" instead of "element.text" to
2176 # keep mypy happy.
2177 self._href = str(element.text).strip()
2178 else:
2179 raise ValueError(
2180 "Property 'href' expected but not found in PROPFIND response: "
2181 f"{eTree.tostring(response, encoding='unicode')}"
2182 )
2184 for propstat in response.findall("./{DAV:}propstat"):
2185 # Only extract properties of interest with status OK.
2186 status = propstat.find("./{DAV:}status")
2187 if status is None or not self._status_ok_rex.match(str(status.text)):
2188 continue
2190 for prop in propstat.findall("./{DAV:}prop"):
2191 # Parse "collection".
2192 if (element := prop.find("./{DAV:}resourcetype/{DAV:}collection")) is not None:
2193 self._collection = True
2195 # Parse "getlastmodified".
2196 if (element := prop.find("./{DAV:}getlastmodified")) is not None:
2197 self._getlastmodified = str(element.text)
2199 # Parse "getcontentlength".
2200 if (element := prop.find("./{DAV:}getcontentlength")) is not None:
2201 self._getcontentlength = int(str(element.text))
2203 # Parse "displayname".
2204 if (element := prop.find("./{DAV:}displayname")) is not None:
2205 self._displayname = str(element.text)
2207 # Some webDAV servers don't include the 'displayname' property in the
2208 # response so try to infer it from the value of the 'href' property.
2209 # Depending on the server the href value may end with '/'.
2210 if not self._displayname:
2211 self._displayname = os.path.basename(self._href.rstrip("/"))
2213 # Force a size of 0 for collections.
2214 if self._collection:
2215 self._getcontentlength = 0
2217 @property
2218 def exists(self) -> bool:
2219 # It is either a directory or a file with length of at least zero
2220 return self._collection or self._getcontentlength >= 0
2222 @property
2223 def is_directory(self) -> bool:
2224 return self._collection
2226 @property
2227 def is_file(self) -> bool:
2228 return not self._collection
2230 @property
2231 def size(self) -> int:
2232 return self._getcontentlength
2234 @property
2235 def name(self) -> str:
2236 return self._displayname
2238 @property
2239 def href(self) -> str:
2240 return self._href
2243class _SessionWrapper(contextlib.AbstractContextManager):
2244 """Wraps a `requests.Session` to allow header values to be injected with
2245 all requests.
2247 Notes
2248 -----
2249 `requests.Session` already has a feature for setting headers globally, but
2250 our session objects are global and authorization headers can vary for each
2251 HttpResourcePath instance.
2252 """
2254 def __init__(self, session: requests.Session, *, extra_headers: dict[str, str] | None) -> None:
2255 self._session = session
2256 self._extra_headers = extra_headers
2258 def __enter__(self) -> _SessionWrapper:
2259 self._session.__enter__()
2260 return self
2262 def __exit__(
2263 self,
2264 exc_type: Any,
2265 exc_value: Any,
2266 traceback: Any,
2267 ) -> None:
2268 return self._session.__exit__(exc_type, exc_value, traceback)
2270 def get(
2271 self,
2272 url: str,
2273 *,
2274 timeout: tuple[float, float],
2275 allow_redirects: bool = True,
2276 stream: bool,
2277 headers: dict[str, str] | None = None,
2278 ) -> requests.Response:
2279 return self._session.get(
2280 url,
2281 timeout=timeout,
2282 allow_redirects=allow_redirects,
2283 stream=stream,
2284 headers=self._augment_headers(headers),
2285 )
2287 def head(
2288 self,
2289 url: str,
2290 *,
2291 timeout: tuple[float, float],
2292 allow_redirects: bool,
2293 stream: bool,
2294 headers: dict[str, str] | None = None,
2295 ) -> requests.Response:
2296 return self._session.head(
2297 url,
2298 timeout=timeout,
2299 allow_redirects=allow_redirects,
2300 stream=stream,
2301 headers=self._augment_headers(headers),
2302 )
2304 def request(
2305 self,
2306 method: str,
2307 url: str,
2308 *,
2309 data: str | bytes | BinaryIO | None,
2310 timeout: tuple[float, float],
2311 allow_redirects: bool,
2312 stream: bool,
2313 headers: dict[str, str] | None = None,
2314 ) -> requests.Response:
2315 return self._session.request(
2316 method,
2317 url,
2318 data=data,
2319 timeout=timeout,
2320 allow_redirects=allow_redirects,
2321 stream=stream,
2322 headers=self._augment_headers(headers),
2323 )
2325 def _augment_headers(self, headers: dict[str, str] | None) -> dict[str, str]:
2326 if headers is None:
2327 headers = {}
2329 if self._extra_headers is not None:
2330 headers = headers | self._extra_headers
2332 return headers