Skip to content

Signing

Signing module.

Revamp of Microsoft Planetary Computer SAS, using custom URL signing endpoint instead.

ExpiredSignedURL

Bases: Exception

The signed URL has expired.

Source code in teledetection/sdk/signing.py 
49
50
class ExpiredSignedURL(Exception):
    """The signed URL has expired."""

NotSignedURL

Bases: Exception

The URL is not signed.

Source code in teledetection/sdk/signing.py 
45
46
class NotSignedURL(Exception):
    """The URL is not signed."""

SignURLRoute

Bases: Enum

Different routes used for sign_urls.

Source code in teledetection/sdk/signing.py 
178
179
180
181
182
class SignURLRoute(Enum):
    """Different routes used for sign_urls."""

    SIGN_URLS_GET = "sign_urls"
    SIGN_URLS_PUT = "sign_urls_put"

SignedURL

Bases: URLBase

Signed URL response.

Source code in teledetection/sdk/signing.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class SignedURL(URLBase):  # pylint: disable = R0903
    """Signed URL response."""

    href: str

    def ttl(self) -> float:
        """Return the number of seconds the token is still valid for."""
        return (self.expiry - datetime.now(timezone.utc)).total_seconds()

    @classmethod
    def from_already_signed(cls, signed_href: str):
        """Create an instance from an already signed URL."""
        parsed_url = urlparse(signed_href.rstrip("/"))
        parsed_qs = parse_qs(parsed_url.query)
        parsed_date = parsed_qs.get("X-Amz-Date", [""])
        parsed_expiry = parsed_qs.get("X-Amz-Expires", [""])

        # Grab date
        try:
            log.debug("Parsing date from signed URL %s (%s)", signed_href, parsed_date)
            parsed_datetime = datetime.strptime(
                parsed_date[0], "%Y%m%dT%H%M%SZ"
            ).replace(tzinfo=timezone.utc)
        except ValueError as err:
            raise NotSignedURL(f"Cannot parse date from URL {signed_href}") from err

        # Grab expiry
        try:
            log.debug("Parsing expiry from signed URL %s", signed_href)
            parsed_expiry_td = timedelta(seconds=int(parsed_expiry[0]))
        except ValueError as err:
            raise NotSignedURL(f"Cannot parse expiry from URL {signed_href}") from err

        # Try to instantiate the SignedURL object and check its TTL
        if (
            url := cls(expiry=parsed_datetime + parsed_expiry_td, href=signed_href)
        ).ttl() < ENV.tld_ttl_margin:
            raise ExpiredSignedURL(f"The signed URL {signed_href} has expired")
        return url

from_already_signed(signed_href) classmethod

Create an instance from an already signed URL.

Source code in teledetection/sdk/signing.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@classmethod
def from_already_signed(cls, signed_href: str):
    """Create an instance from an already signed URL."""
    parsed_url = urlparse(signed_href.rstrip("/"))
    parsed_qs = parse_qs(parsed_url.query)
    parsed_date = parsed_qs.get("X-Amz-Date", [""])
    parsed_expiry = parsed_qs.get("X-Amz-Expires", [""])

    # Grab date
    try:
        log.debug("Parsing date from signed URL %s (%s)", signed_href, parsed_date)
        parsed_datetime = datetime.strptime(
            parsed_date[0], "%Y%m%dT%H%M%SZ"
        ).replace(tzinfo=timezone.utc)
    except ValueError as err:
        raise NotSignedURL(f"Cannot parse date from URL {signed_href}") from err

    # Grab expiry
    try:
        log.debug("Parsing expiry from signed URL %s", signed_href)
        parsed_expiry_td = timedelta(seconds=int(parsed_expiry[0]))
    except ValueError as err:
        raise NotSignedURL(f"Cannot parse expiry from URL {signed_href}") from err

    # Try to instantiate the SignedURL object and check its TTL
    if (
        url := cls(expiry=parsed_datetime + parsed_expiry_td, href=signed_href)
    ).ttl() < ENV.tld_ttl_margin:
        raise ExpiredSignedURL(f"The signed URL {signed_href} has expired")
    return url

ttl()

Return the number of seconds the token is still valid for.

Source code in teledetection/sdk/signing.py 
65
66
67
def ttl(self) -> float:
    """Return the number of seconds the token is still valid for."""
    return (self.expiry - datetime.now(timezone.utc)).total_seconds()

SignedURLBatch

Bases: URLBase

Signed URLs (batch of URLs) response.

Source code in teledetection/sdk/signing.py 
101
102
103
104
class SignedURLBatch(URLBase):  # pylint: disable = R0903
    """Signed URLs (batch of URLs) response."""

    hrefs: dict

URLBase

Bases: BaseModel

Base model for responses.

Source code in teledetection/sdk/signing.py 
53
54
55
56
57
class URLBase(BaseModel):  # pylint: disable = R0903
    """Base model for responses."""

    model_config = ConfigDict(populate_by_name=True)
    expiry: datetime

is_vrt_string(string)

Check whether a string looks like a VRT.

Source code in teledetection/sdk/signing.py 
143
144
145
146
147
def is_vrt_string(string: str) -> bool:
    """Check whether a string looks like a VRT."""
    return string.strip().startswith("<VRTDataset") and string.strip().endswith(
        "</VRTDataset>"
    )

sign(obj, copy=True)

Sign the relevant URL with a S3 token allowing read access.

All URLs belonging to supported objects are modified in-place, or returned by the function, depending on copy. Args: obj (Any): The object to sign. Must be one of: str (URL), Asset, Item, ItemCollection, or ItemSearch, or a mapping. copy (bool): Whether to sign the object in place, or make a copy. Has no effect for immutable objects like strings.

Returns:

Name Type Description
Any Any

A copy of the object where all relevant URLs have been signed
Source code in teledetection/sdk/signing.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@singledispatch
def sign(obj: Any, copy: bool = True) -> Any:
    """Sign the relevant URL with a S3 token allowing read access.

    All URLs belonging to supported objects are modified in-place, or returned
    by the function, depending on `copy`.
    Args:
        obj (Any): The object to sign. Must be one of:
            str (URL), Asset, Item, ItemCollection, or ItemSearch, or a
            mapping.
        copy (bool): Whether to sign the object in place, or make a copy.
            Has no effect for immutable objects like strings.

    Returns:
        Any: A copy of the object where all relevant URLs have been signed

    """
    raise TypeError(
        "Invalid type, must be one of: str, Asset, Item, ItemCollection, ItemSearch, or mapping"
    )

sign_asset(asset, copy=True)

Sign a PySTAC asset.

Parameters:

Name Type Description Default
asset Asset

The Asset to sign

 required
copy bool

Whether to copy (clone) the asset or mutate it inplace.

 True

Returns:

Name Type Description
Asset Asset

An asset where the HREF is replaced with a
Asset

signed version.
Source code in teledetection/sdk/signing.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
@sign.register(Asset)
def sign_asset(asset: Asset, copy: bool = True) -> Asset:
    """Sign a PySTAC asset.

    Args:
        asset (Asset): The Asset to sign
        copy (bool): Whether to copy (clone) the asset or mutate it inplace.

    Returns:
        Asset: An asset where the HREF is replaced with a
        signed version.

    """
    if copy:
        asset = asset.clone()
    asset.href = sign_urls([asset.href])[asset.href]
    return asset

sign_collection(collection, copy=True)

Sign a collection.

Parameters:

Name Type Description Default
collection Collection

STAC Collection

 required
copy bool

copy or not the input

 True

Returns:

Name Type Description
signed Collection

the STAC collection, now with signed URLs.
Source code in teledetection/sdk/signing.py 
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
@sign.register(Collection)
def sign_collection(collection: Collection, copy: bool = True) -> Collection:
    """Sign a collection.

    Args:
        collection: STAC Collection
        copy: copy or not the input

    Returns:
        signed (Collection): the STAC collection, now with signed URLs.

    """
    if copy:
        # https://github.com/stac-utils/pystac/pull/834 fixed asset dropping
        assets = collection.assets
        collection = collection.clone()
        if assets and not collection.assets:
            collection.assets = deepcopy(assets)

    urls = [collection.assets[key].href for key in collection.assets]
    signed_urls = sign_urls(urls=urls)
    for key, asset in collection.assets.items():
        collection.assets[key].href = signed_urls[asset.href]
    return collection

sign_inplace(obj)

Sign the object in place.

See :func:teledetection.sign for more.

Source code in teledetection/sdk/signing.py 
134
135
136
137
138
139
140
def sign_inplace(obj: Any) -> Any:
    """Sign the object in place.

    See :func:`teledetection.sign` for more.

    """
    return sign(obj, copy=False)

sign_item(item, copy=True)

Sign all assets within a PySTAC item.

Parameters:

Name Type Description Default
item Item

The Item whose assets that will be signed

 required
copy bool

Whether to copy (clone) the item or mutate it inplace.

 True

Returns:

Name Type Description
Item Item

An Item where all assets' HREFs have
Item

been replaced with a signed version. In addition, an "expiry"
Item

property is added to the Item properties indicating the earliest
Item

expiry time for any assets that were signed.
Source code in teledetection/sdk/signing.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
@sign.register(Item)
def sign_item(item: Item, copy: bool = True) -> Item:
    """Sign all assets within a PySTAC item.

    Args:
        item (Item): The Item whose assets that will be signed
        copy (bool): Whether to copy (clone) the item or mutate it inplace.

    Returns:
        Item: An Item where all assets' HREFs have
        been replaced with a signed version. In addition, an "expiry"
        property is added to the Item properties indicating the earliest
        expiry time for any assets that were signed.

    """
    if copy:
        item = item.clone()
    urls = [asset.href for asset in item.assets.values()]
    signed_urls = sign_urls(urls=urls)
    for key, asset in item.assets.items():
        item.assets[key].href = signed_urls[asset.href]
    return item

sign_item_collection(item_collection, copy=True)

Sign a PySTAC item collection.

Parameters:

Name Type Description Default
item_collection ItemCollection

The ItemCollection whose assets will be signed

 required
copy bool

Whether to copy (clone) the ItemCollection or mutate it inplace.

 True

Returns:

Name Type Description
ItemCollection ItemCollection

An ItemCollection where all assets'
ItemCollection

HREFs for each item have been replaced with a signed version. In
ItemCollection

addition, an "expiry" property is added to the Item properties
ItemCollection

indicating the earliest expiry time for any assets that were signed.
Source code in teledetection/sdk/signing.py 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
@sign.register(ItemCollection)
def sign_item_collection(
    item_collection: ItemCollection, copy: bool = True
) -> ItemCollection:
    """Sign a PySTAC item collection.

    Args:
        item_collection (ItemCollection): The ItemCollection whose assets will
            be signed
        copy (bool): Whether to copy (clone) the ItemCollection or mutate it
            inplace.

    Returns:
        ItemCollection: An ItemCollection where all assets'
        HREFs for each item have been replaced with a signed version. In
        addition, an "expiry" property is added to the Item properties
        indicating the earliest expiry time for any assets that were signed.

    """
    if copy:
        item_collection = item_collection.clone()
    urls = [asset.href for item in item_collection for asset in item.assets.values()]
    signed_urls = sign_urls(urls=urls)
    for item in item_collection:
        for key, asset in item.assets.items():
            item.assets[key].href = signed_urls[asset.href]
    return item_collection

sign_mapping(mapping, copy=True)

Sign a mapping.

Parameters:

Name Type Description Default
mapping Mapping
required
copy bool

Whether to copy (clone) the mapping or mutate it inplace.

 True

Returns:

Name Type Description
signed Mapping

The dictionary, now with signed URLs.
Source code in teledetection/sdk/signing.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
@sign.register(collections.abc.Mapping)
def sign_mapping(mapping: Mapping, copy: bool = True) -> Mapping:
    """Sign a mapping.

    Args:
        mapping (Mapping):

        The mapping (e.g. dictionary) to sign. This method can sign

            * Kerchunk-style references, which signs all URLs under the
              ``templates`` key. See https://fsspec.github.io/kerchunk/
              for more.
            * STAC items
            * STAC collections
            * STAC ItemCollections

        copy: Whether to copy (clone) the mapping or mutate it inplace.

    Returns:
        signed (Mapping): The dictionary, now with signed URLs.

    """
    if copy:
        mapping = deepcopy(mapping)

    types = (STACObjectType.ITEM, STACObjectType.COLLECTION)
    if all(key in mapping for key in ["version", "templates", "refs"]):
        urls = list(mapping["templates"].values())
        signed_urls = sign_urls(urls=urls)
        for key, url in mapping["templates"].items():
            mapping["templates"][key] = signed_urls[url]

    elif identify_stac_object_type(cast(Dict[str, Any], mapping)) in types:
        urls = [val["href"] for val in mapping["assets"].values()]
        signed_urls = sign_urls(urls=urls)
        for val in mapping["assets"].values():
            url = val["href"]
            val["href"] = signed_urls[url]

    elif mapping.get("type") == "FeatureCollection" and mapping.get("features"):
        urls = [
            val["href"]
            for feat in mapping["features"]
            for val in feat.get("assets", {}).values()
        ]
        signed_urls = sign_urls(urls=urls)
        for feature in mapping["features"]:
            for val in feature.get("assets", {}).values():
                url = val["href"]
                val["href"] = signed_urls[url]

    return mapping

sign_string(url, copy=True)

Sign a URL or VRT-like string containing URLs with a S3 Token.

Signing with a S3 token allows read access to files in blob storage.

Parameters:

Name Type Description Default
url str

The HREF of the asset as a URL or a GDAL VRT

Single URLs can be found on a STAC Item's Asset href value. Only URLs to assets in S3 Storage are signed, other URLs are returned unmodified.

GDAL VRTs can combine many data sources into a single mosaic. A VRT can be built quickly from the GDAL STACIT driver https://gdal.org/drivers/raster/stacit.html. Each URL to S3 Storage within the VRT is signed.

 required
copy bool

No effect.

 True

Returns:

Name Type Description
str str

The signed HREF or VRT
Source code in teledetection/sdk/signing.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@sign.register(str)
def sign_string(url: str, copy: bool = True) -> str:
    """Sign a URL or VRT-like string containing URLs with a S3 Token.

    Signing with a S3 token allows read access to files in blob storage.

    Args:
        url (str): The HREF of the asset as a URL or a GDAL VRT

            Single URLs can be found on a STAC Item's Asset ``href`` value.
            Only URLs to assets in S3 Storage are signed, other URLs are
            returned unmodified.

            GDAL VRTs can combine many data sources into a single mosaic.
            A VRT can be  built quickly from the GDAL STACIT driver
            https://gdal.org/drivers/raster/stacit.html. Each URL to S3 Storage
            within the VRT is signed.
        copy (bool): No effect.

    Returns:
        str: The signed HREF or VRT

    """
    if is_vrt_string(url):
        return sign_vrt_string(url)
    return sign_urls(urls=[url])[url]

sign_url_put(url)

Sign a single URL for PUT.

Source code in teledetection/sdk/signing.py 
233
234
235
236
def sign_url_put(url: str) -> str:
    """Sign a single URL for PUT."""
    urls = sign_urls_put([url])
    return urls[url]

sign_urls(urls)

Sign multiple URLs for GET.

Source code in teledetection/sdk/signing.py 
223
224
225
def sign_urls(urls: list[str]) -> Dict[str, str]:
    """Sign multiple URLs for GET."""
    return _generic_sign_urls(urls=urls, route=SignURLRoute(SignURLRoute.SIGN_URLS_GET))

sign_urls_put(urls)

Sign multiple URLs for PUT.

Source code in teledetection/sdk/signing.py 
228
229
230
def sign_urls_put(urls: list[str]) -> Dict[str, str]:
    """Sign multiple URLs for PUT."""
    return _generic_sign_urls(urls=urls, route=SignURLRoute(SignURLRoute.SIGN_URLS_PUT))

sign_vrt_string(vrt, copy=True)

Sign a VRT-like string containing URLs from the storage.

Signing URLs allows read access to files in storage.

Parameters:

Name Type Description Default
vrt str

The GDAL VRT

GDAL VRTs can combine many data sources into a single mosaic. A VRT can be built quickly from the GDAL STACIT driver https://gdal.org/drivers/raster/stacit.html. Each URL to S3 Storage within the VRT is signed.

 required
copy bool

No effect.

 True

Returns:

Name Type Description
str str

The signed VRT
Source code in teledetection/sdk/signing.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def sign_vrt_string(vrt: str, copy: bool = True) -> str:  # pylint: disable = W0613  # noqa: E501
    """Sign a VRT-like string containing URLs from the storage.

    Signing URLs allows read access to files in storage.

    Args:
        vrt (str): The GDAL VRT

            GDAL VRTs can combine many data sources into a single mosaic. A VRT
            can be built quickly from the GDAL STACIT driver
            https://gdal.org/drivers/raster/stacit.html. Each URL to S3 Storage
            within the VRT is signed.
        copy (bool): No effect.

    Returns:
        str: The signed VRT

    """
    urls = []

    def _repl_vrt(m: re.Match):
        urls.append(m.string[slice(*m.span())])

    asset_xpr.sub(_repl_vrt, vrt)
    signed_urls = sign_urls(urls)

    # The "&" needs to be encoded in signed URLs inside the .vrt
    for url, signed_url in signed_urls.items():
        signed_urls[url] = signed_url.replace("&", "&Amp;")

    # Replace urls with signed urls (single pass)
    rep = dict((re.escape(k), v) for k, v in signed_urls.items())
    pattern = re.compile("|".join(rep.keys()))
    return pattern.sub(lambda m: rep[re.escape(m.group(0))], vrt)