Source code for digital_world_sdk.sep.dw_toml

"""
DWP: 0001
Title: dw.toml
Author: digitalworld.global
Status: Active
Created: 2017-10-30
Updated: 2019-06-12
Version: 2.1.0
"""

import sys
from collections.abc import MutableMapping
from typing import Any

if sys.version_info >= (3, 11):
    from tomllib import loads as toml_loads
else:
    from toml import loads as toml_loads

from ..client.aiohttp_client import AiohttpClient
from ..client.base_async_client import BaseAsyncClient
from ..client.base_sync_client import BaseSyncClient
from ..client.requests_client import RequestsClient
from ..client.response import Response
from .exceptions import DwTomlNotFoundError

__all__ = ["fetch_dw_toml", "fetch_dw_toml_async"]

# Maximum allowed size for dw.toml file (512 KB).
# This limit helps prevent denial-of-service attacks via memory exhaustion.
DW_TOML_MAX_SIZE = 512 * 1024




def fetch_dw_toml(
    domain: str,
    client: BaseSyncClient | None = None,
    use_http: bool = False,
) -> MutableMapping[str, Any]:
    """Retrieve the dw.toml file from a given domain.

    Retrieve the dw.toml file for information about interacting with
    Digital World federation protocol for a given Digital World Anchor (specified by a
    domain).

    :param domain: The domain the .toml file is hosted at.
    :param use_http: Specifies whether the request should go over plain HTTP vs HTTPS.
        Note it is recommended that you **always** use HTTPS.
    :param client: Http Client used to send the request.
    :return: The dw.toml file as an object via :func:`toml.loads`.
    :raises: :exc:`DwTomlNotFoundError <digital_world_sdk.sep.exceptions.DwTomlNotFoundError>`:
        if the Digital World toml file could not be found.
    :raises: :exc:`ContentSizeLimitExceededError <digital_world_sdk.exceptions.ContentSizeLimitExceededError>`:
        if the response size exceeds the maximum allowed size.
    """
    if not client:
        client = RequestsClient()
    url = _build_request_url(domain, use_http)
    raw_resp = client.get(url, max_content_size=DW_TOML_MAX_SIZE)
    return _handle_raw_response(raw_resp)





async def fetch_dw_toml_async(
    domain: str,
    client: BaseAsyncClient | None = None,
    use_http: bool = False,
) -> MutableMapping[str, Any]:
    """Retrieve the dw.toml file from a given domain.

    Retrieve the dw.toml file for information about interacting with
    Digital World federation protocol for a given Digital World Anchor (specified by a
    domain).

    :param domain: The domain the .toml file is hosted at.
    :param use_http: Specifies whether the request should go over plain HTTP vs HTTPS.
        Note it is recommended that you **always** use HTTPS.
    :param client: Http client used to send the request.
    :return: The dw.toml file as a dict object.
    :raises: :exc:`DwTomlNotFoundError <digital_world_sdk.sep.exceptions.DwTomlNotFoundError>`:
        if the Digital World toml file could not be found.
    :raises: :exc:`ContentSizeLimitExceededError <digital_world_sdk.exceptions.ContentSizeLimitExceededError>`:
        if the response size exceeds the maximum allowed size.
    """

    if not client:
        client = AiohttpClient()
    url = _build_request_url(domain, use_http)
    raw_resp = await client.get(url, max_content_size=DW_TOML_MAX_SIZE)
    return _handle_raw_response(raw_resp)



def _handle_raw_response(raw_resp: Response) -> MutableMapping[str, Any]:
    if raw_resp.status_code == 404:
        raise DwTomlNotFoundError
    resp = raw_resp.text
    return toml_loads(resp)


def _build_request_url(domain: str, use_http: bool = False) -> str:
    toml_link = "/.well-known/dw.toml"
    protocol = "https://"
    if use_http:
        protocol = "http://"
    url = protocol + domain + toml_link
    return url