"""
DWP: 0002
Title: Federation protocol
Author: digitalworld.global
Status: Final
Created: 2017-10-30
Updated: 2019-10-10
Version 1.1.0
"""
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 (
BadFederationResponseError,
FederationServerNotFoundError,
InvalidFederationAddress,
)
from .dw_toml import fetch_dw_toml, fetch_dw_toml_async
SEPARATOR = "*"
FEDERATION_SERVER_KEY = "FEDERATION_SERVER"
# Maximum allowed size for federation response (100 KB).
# This limit helps prevent denial-of-service attacks via memory exhaustion.
FEDERATION_RESPONSE_MAX_SIZE = 100 * 1024
__all__ = [
"FederationRecord",
"resolve_account_id",
"resolve_account_id_async",
"resolve_dw_address",
"resolve_dw_address_async",
]
[docs]
class FederationRecord:
def __init__(
self,
account_id: str,
dw_address: str,
memo_type: str | None,
memo: str | None,
) -> None:
"""The :class:`FederationRecord`, which represents record in federation server.
:param account_id: Digital World public key / account ID
:param dw_address: Digital World address
:param memo_type: Type of memo to attach to transaction, one of ``text``, ``id`` or ``hash``
:param memo: value of memo to attach to transaction, for ``hash`` this should be base64-encoded.
This field should always be of type ``string`` (even when `memo_type` is equal ``id``) to support parsing
value in languages that don't support big numbers.
"""
self.account_id: str = account_id
self.dw_address: str = dw_address
self.memo_type: str | None = memo_type
self.memo: str | None = memo
def __hash__(self):
return hash((self.account_id, self.dw_address, self.memo_type, self.memo))
def __repr__(self):
return (
f"<FederationRecord [account_id={self.account_id}, dw_address={self.dw_address}, "
f"memo_type={self.memo_type}, memo={self.memo}]>"
)
def __eq__(self, other: object) -> bool:
if not isinstance(other, self.__class__):
return NotImplemented
return (
self.account_id == other.account_id
and self.dw_address == other.dw_address
and self.memo_type == other.memo_type
and self.memo == other.memo
)
[docs]
def resolve_dw_address(
dw_address: str,
client: BaseSyncClient | None = None,
federation_url: str | None = None,
use_http: bool = False,
) -> FederationRecord:
"""Get the federation record if the user was found for a given Digital World address.
:param dw_address: Digital World address (ex. ``"bob*digitalworld.global"``).
:param client: Http Client used to send the request.
:param federation_url: The federation server URL (ex. ``"https://digitalworld.global/federation"``),
if you don't set this value, we will try to get it from `dw_address`.
:param use_http: Specifies whether the request should go over plain HTTP vs HTTPS.
Note it is recommended that you **always** use HTTPS.
:return: Federation record.
"""
if not client:
client = RequestsClient()
parts = _split_dw_address(dw_address)
domain = parts["domain"]
if federation_url is None:
federation_url = fetch_dw_toml(domain, use_http=use_http).get(
FEDERATION_SERVER_KEY
)
if federation_url is None:
raise FederationServerNotFoundError(
f"Unable to find federation server at {domain}."
)
raw_resp = client.get(
federation_url,
{"type": "name", "q": dw_address},
max_content_size=FEDERATION_RESPONSE_MAX_SIZE,
)
return _handle_raw_response(raw_resp, dw_address=dw_address)
[docs]
async def resolve_dw_address_async(
dw_address: str,
client: BaseAsyncClient | None = None,
federation_url: str | None = None,
use_http: bool = False,
) -> FederationRecord:
"""Get the federation record if the user was found for a given Digital World address.
:param dw_address: Digital World address (ex. ``"bob*digitalworld.global"``).
:param client: Http Client used to send the request.
:param federation_url: The federation server URL (ex. ``"https://digitalworld.global/federation"``),
if you don't set this value, we will try to get it from `dw_address`.
:param use_http: Specifies whether the request should go over plain HTTP vs HTTPS.
Note it is recommended that you **always** use HTTPS.
:return: Federation record.
"""
if not client:
client = AiohttpClient()
parts = _split_dw_address(dw_address)
domain = parts["domain"]
if federation_url is None:
federation_url = (
await fetch_dw_toml_async(domain, client=client, use_http=use_http)
).get(FEDERATION_SERVER_KEY)
if federation_url is None:
raise FederationServerNotFoundError(
f"Unable to find federation server at {domain}."
)
raw_resp = await client.get(
federation_url,
{"type": "name", "q": dw_address},
max_content_size=FEDERATION_RESPONSE_MAX_SIZE,
)
return _handle_raw_response(raw_resp, dw_address=dw_address)
[docs]
def resolve_account_id(
account_id: str,
domain: str | None = None,
federation_url: str | None = None,
client: BaseSyncClient | None = None,
use_http: bool = False,
) -> FederationRecord:
"""Given an account ID, get their federation record if the user was found
:param account_id: Account ID (ex. ``"GBYNR2QJXLBCBTRN44MRORCMI4YO7FZPFBCNOKTOBCAAFC7KC3LNPRYS"``)
:param domain: Get `federation_url` from the domain, you don't need to set this value if `federation_url` is set.
:param federation_url: The federation server URL (ex. ``"https://digitalworld.global/federation"``).
:param client: Http Client used to send the request.
:param use_http: Specifies whether the request should go over plain HTTP vs HTTPS.
Note it is recommended that you **always** use HTTPS.
:return: Federation record.
"""
if domain is None and federation_url is None:
raise ValueError("You should provide either `domain` or `federation_url`.")
if not client:
client = RequestsClient()
if domain is not None:
federation_url = fetch_dw_toml(domain, client, use_http).get(
FEDERATION_SERVER_KEY
)
if federation_url is None:
raise FederationServerNotFoundError(
f"Unable to find federation server at {domain}."
)
assert federation_url is not None
raw_resp = client.get(
federation_url,
{"type": "id", "q": account_id},
max_content_size=FEDERATION_RESPONSE_MAX_SIZE,
)
return _handle_raw_response(raw_resp, account_id=account_id)
[docs]
async def resolve_account_id_async(
account_id: str,
domain: str | None = None,
federation_url: str | None = None,
client: BaseAsyncClient | None = None,
use_http: bool = False,
) -> FederationRecord:
"""Given an account ID, get their federation record if the user was found
:param account_id: Account ID (ex. ``"GBYNR2QJXLBCBTRN44MRORCMI4YO7FZPFBCNOKTOBCAAFC7KC3LNPRYS"``)
:param domain: Get `federation_url` from the domain, you don't need to set this value if `federation_url` is set.
:param federation_url: The federation server URL (ex. ``"https://digitalworld.global/federation"``).
:param client: Http Client used to send the request.
:param use_http: Specifies whether the request should go over plain HTTP vs HTTPS.
Note it is recommended that you **always** use HTTPS.
:return: Federation record.
"""
if domain is None and federation_url is None:
raise ValueError("You should provide either `domain` or `federation_url`.")
if not client:
client = AiohttpClient()
if domain is not None:
federation_url = (await fetch_dw_toml_async(domain, client, use_http)).get(
FEDERATION_SERVER_KEY
)
if federation_url is None:
raise FederationServerNotFoundError(
f"Unable to find federation server at {domain}."
)
assert federation_url is not None
raw_resp = await client.get(
federation_url,
{"type": "id", "q": account_id},
max_content_size=FEDERATION_RESPONSE_MAX_SIZE,
)
return _handle_raw_response(raw_resp, account_id=account_id)
def _handle_raw_response(
raw_resp: Response, dw_address=None, account_id=None
) -> FederationRecord:
if not 200 <= raw_resp.status_code < 300:
raise BadFederationResponseError(raw_resp)
data = raw_resp.json()
account_id = account_id or data.get("account_id")
# "digital_world_address" is the wire-format field name for DWP-0002 federation
# responses served by Digital World Gateway/federation servers. This is not a
# third-party wire constant we must match for interop with an external network -
# Digital World controls both the federation server and this client, and the
# JS SDK's reference implementation already uses "digital_world_address" for the
# same field (see test/unit/federation_server.test.ts) - so it is intentionally
# renamed here to match, rather than left as the legacy Stellar SEP-0002 name.
dw_address = dw_address or data.get("digital_world_address")
memo_type = data.get("memo_type")
memo = data.get("memo")
if not account_id or not dw_address:
# This should never happen, but if the server-side implementation is not standardized, this error may be triggered.
raise BadFederationResponseError(raw_resp)
return FederationRecord(
account_id=account_id,
dw_address=dw_address,
memo_type=memo_type,
memo=memo,
)
def _split_dw_address(address: str) -> dict[str, str]:
parts = address.split(SEPARATOR)
if len(parts) != 2:
raise InvalidFederationAddress(
"Address should be a valid address, such as `bob*digitalworld.global`"
)
name, domain = parts
return {"name": name, "domain": domain}