PyFederation

PyFederation is a Python client library for the Federated Database protocol. It implements the Open Federated Database specification and follows the design of the reference implementation, FederationLib, and of its Java counterpart, JFederation, from which it is ported.

Features

• Works with any Federation server implementing OFD 2.x
• Content scanning with entity resolution, classification, risk assessment, content types, concerns, detections, and files such as the images of a message
• Every method of the server API: operators, records, evidence, reports, blacklist records, file attachments, audit logs, search, and server information
• Pagination, category filters, and sorting on every listing and search endpoint
• Synchronous and asynchronous clients, FederationClient and AsyncFederationClient, with the same methods
• All errors are mapped to a single exception carrying the HTTP status code and server message

Table of Contents

• PyFederation
  • Features
  • Table of Contents
  • Prerequisites
  • Installation
    • Via PyPI
    • From source
  • Usage
    • Connecting to a server
    • Asynchronous client
    • Content scanning
    • Searching records
    • Operators
    • Entities
    • Evidence
    • Reports
    • Blacklist records
    • File attachments
    • Pagination and sorting
  • Error handling
  • Testing
• License

Prerequisites

• Python 3.11 or newer
• A running Federation server for anything other than constructing a client, the reference implementation is FederationLib, which includes a dockerized server deployment

Dependencies are HTTPX for HTTP transport and Pydantic for the records.

Installation

Via PyPI

The library is published on PyPI:

pip install pyfederation

From source

To use an unreleased version, install the library from a clone of the repository:

git clone https://github.com/nosial/PyFederation.git
cd PyFederation
pip install .

Usage

All functionality is exposed through the FederationClient class. Each method maps to a REST endpoint of the Federation server and handles serialization, authentication, and error mapping.

Connecting to a server

A client is constructed with the base URL of the Federation server and an optional operator access token. A client without a token operates anonymously against the server's public endpoints:

from pyfederation import FederationClient

# Anonymous client
client = FederationClient("https://federation.example.com")

# Authenticated client with an operator access token
client = FederationClient("https://federation.example.com", "access-token")

The access token can be changed at any time during the lifetime of the client:

client.access_token = "new-token"  # switch authentication
client.access_token = None  # revert to anonymous access

Clients should be closed when no longer needed to release the underlying HTTP connection pool, or used as a context manager:

client.close()

with FederationClient("https://federation.example.com", "access-token") as client:
    ...

For custom HTTP settings (timeouts, proxies, TLS, etc.) an existing httpx.Client can be passed to the constructor instead:

import httpx

http_client = httpx.Client(timeout=15)

client = FederationClient("https://federation.example.com", "access-token", http_client)

Server information, including the server name, API version, feature flags (which record types are publicly visible), and total record counts, is available through get_server_information(), the raw OpenAPI specification of the server is available through get_specification():

info = client.get_server_information()
print(f"Server {info.name}, API version {info.api_version}")
print(f"Known entities: {info.known_entities}, reports: {info.reports}")

A client should check the major version of the interface with info.api_major_version, this release implements OFD 2.2. Features of a minor version are checked with info.is_api_version_at_least(2, 2) (or info.api_minor_version). The server information also publishes what content scanning supports (scan_content_types, concerns and default_concerns, and the limits on scanned files, max_scan_files and max_scan_file_size), whether URL uploads are accepted (url_uploads_allowed), whether telephone entities are public (public_telephone_entities) and whether content scanned anonymously may be stored (collects_public_scan_content). Values a newer server publishes that this client does not recognize are read as None or left out of lists, and an unrecognized incident type is read as OTHER.

Asynchronous client

AsyncFederationClient has the same methods as FederationClient, as coroutines, and is closed with aclose() or used as an async context manager:

from pyfederation import AsyncFederationClient

async with AsyncFederationClient("https://federation.example.com", "access-token") as client:
    info = await client.get_server_information()

Routes the client has no method for, such as the routes a plugin of the server provides, are reached with request(), which sends the parameters as the query string of a GET or DELETE request and as a JSON body otherwise, and returns the decoded response:

status = client.request("GET", "my-plugin/status", {"verbose": "true"})

Content scanning

Content scanning accepts text, a ContentInput, or a list of inputs. Each input can carry text, a content type and its context, an operator note, tag, confidentiality flag, metadata, and files; at least one input must contain text or a file:

The content type (TEXT, EMAIL, SMS, WEB_PAGE, HOSTNAME or URL) determines the form of the text and the keys its context may contain, such as subject and sender for an EMAIL, region for an SMS, and the required url for a WEB_PAGE. The context is an assertion of the client, a detection that is not verified relies on it. Only the requested concerns (the server's default_concerns when none are given, which exclude ADVERTISING and TRACKING) affect the classification, risk score, suggested action, and detections of the result:

from pyfederation import ContentInput, ContentType, IncidentType

scan = client.scan_content(
    [
        ContentInput(
            "Your account is locked, verify it now",
            content_type=ContentType.EMAIL,
            context={"subject": "Account locked", "sender": "security@example.com"},
        ),
        ContentInput("https://login.example.com/verify", content_type=ContentType.URL),
    ],
    concerns=[IncidentType.PHISHING, IncidentType.SPAM],
)

for detection in scan.detections:
    print(
        f"{detection.concern} in input {detection.content_index} by {detection.source} "
        f"(verified: {detection.verified})"
    )

The files of each input, such as the images of a message, are given as local paths keyed by the position of the input, and an input with files may omit its text. The server determines the media type of each file from its content, interprets the ones it supports, and ignores the others, as it ignores every file when max_scan_files is 0; servers before OFD 2.1 may reject files. A detection that relates to a file carries its position among the files of the input in file_index:

scan = client.scan_content(
    [
        ContentInput("Your parcel is on hold", content_type=ContentType.SMS),
        ContentInput(),  # an image sent on its own
    ],
    files={0: ["parcel.jpg"], 1: ["qr.png"]},
)

Files the server records as evidence are kept as attachments of that evidence, and many files carry metadata, such as the location a photograph was taken at, that the server does not need.

Files can also be named by URL, keyed the same way, for the server to retrieve itself, such as the media of an MMS message held by the carrier. A request that only names file URLs is sent as JSON. Servers before OFD 2.2 reject file URLs, and a server that does not retrieve them, or cannot retrieve a file, treats it like a file it does not interpret, so the result does not depend on whether it does. Retrieved files follow the files of their input in file_index:

if info.is_api_version_at_least(2, 2):
    scan = client.scan_content(
        ContentInput(content_type=ContentType.SMS),
        file_urls={0: ["https://carrier.example/mms/1.jpg"]},
    )

Scanning content and inspecting the classification result:

from pyfederation import ContentInput

scan = client.scan_content(ContentInput("Suspicious message to inspect"), "author@example.com")

classification = scan.classification
print(
    f"Classification {classification.classification_flag} with "
    f"{classification.confidence:.1%} confidence, detected language: {classification.detected_language}"
)

Searching records

The global search queries all record types with an optional type filter. The search query must contain at least 2 non-whitespace characters:

from pyfederation import RecordType

# Search all record types
results = client.search("author@example.com")

# Search only entity and blacklist records, page 1, 25 per page
results = client.search("author@example.com", [RecordType.ENTITY, RecordType.BLACKLIST], 1, 25)

SearchResult records carry the matched type and the record, resolved to the record class of the matched type.

Operators

Operators are the accounts of the Federation server. Every change to the data is made by an operator, and each operator holds permissions that decide what it may do.

Creating an operator and generating a new token:

created = client.create_operator("my-application")
print(f"Operator UUID: {created.uuid}")

# The raw token is only returned once at creation time, store it securely
token = created.access_token

refreshed = client.generate_access_token(True)  # refresh and update this client

An OperatorRecord carries its permission groups in permissions and its settings in configuration. Management permissions inherit client permissions, so revoking client permissions from an operator that keeps management permissions is rejected with HTTP 409. The configuration is only disclosed to the operator itself and to operators with operator permissions, anyone else receives None; an operator may change its own configuration:

operator = client.get_operator(uuid)
if operator.permissions.management and operator.configuration is not None:
    print(f"Bot: {operator.is_bot}, auto assigned: {operator.auto_assigned}")

# Opt in to automated reports only
client.update_operator_configuration(uuid, False, True)

Entities

Entities are the subjects tracked by the federation: users, domains, IP addresses, or any other identifier present in scanned content.

Entities are identified by their UUID, SHA-256 hash, or address, such as example.com or john@example.com. Entity identifiers are percent-encoded in request paths, so addresses such as user+tag@example.com can be passed as they are.

A telephone number is the entity <E.164 number>@e164.arpa, whose local part is the international number without + or separators (EntityRecord.is_telephone). Unless the server publishes public_telephone_entities, unauthenticated requests only see a telephone entity they identify by its address or SHA-256 hash.

Evidence

Evidence records are the supporting data attached to entities, either when submitting a report or manually by operators, and can hold text content, attachments, tags, notes, and confidentiality flags:

Confidential evidence is hidden from unauthenticated clients by default, listing endpoints support an include_confidential parameter to include them when the operator has the appropriate permissions. Evidence the server creates itself or a bot operator submits is automated and created confidential, the AUTOMATED and NOT_AUTOMATED categories select it, and evidence about a telephone entity is always confidential.

Reports

Reports are the central mechanism of the federation server. A report is submitted against an entity, groups one or more pieces of evidence, and can be classified, assigned to operators, and closed:

submit_report returns a ReportSubmission object with the created report and its list of evidence records. Upload attachments separately against an evidence record.

Blacklist records

Blacklist records are the server's mitigation mechanism. An operator blacklists an entity with an incident type and, optionally, the report backing the decision, which may concern another entity, such as the domain a reported account advertised:

File attachments

Evidence records support file attachments with uploads from disk or URL, and note attachments (plain-text content):

URL uploads default to a maximum file size of 50 MB, matching the FederationLib client default, the limit can be raised per request.

upload_file_attachment_from_url downloads the file and uploads it, while upload_file_attachment_by_url sends only the URL and the server retrieves the file itself. The URL must then be reachable by the server, and servers reject URLs that resolve to private, loopback or other non-public addresses unless configured otherwise.

Pagination and sorting

Every listing and search endpoint accepts the same pagination and sorting arguments: page (1-based) and limit (minimum 1), followed by an optional category filter, by (the field to sort by, case-insensitive, sent uppercase as OFD 2.0 defines), and order ("ASC" or "DESC"), for example:

# Defaults: page 1, limit 100
entities = client.list_entities()

# Explicit pagination
entities = client.list_entities(2, 50)

# Pagination with a category filter, sorted by reputation ascending
entities = client.list_entities(2, 50, "WHITELISTED", "reputation", "ASC")

The argument order is the same for every endpoint.

Error handling

All failures are raised as FederationClientException, carrying the HTTP status code of the response and, when the server returns one, its error message:

from pyfederation import FederationClientException

try:
    client.get_entity_record("non-existent-entity")
except FederationClientException as e:
    print(f"Request failed with status {e.status_code}: {e}")

A server may limit the rate of requests. A request it rejects with HTTP 429, or with 503 while it is unavailable, was not performed and may be repeated, including one that is not idempotent: e.is_retryable tells so, and e.retry_after holds the delay in seconds the server asked for, or None, in which case back off exponentially. The client does not retry by itself:

import time

for attempt in range(5):
    try:
        scan = client.scan_content("Suspicious message to inspect")
        break
    except FederationClientException as e:
        if not e.is_retryable:
            raise
        time.sleep(e.retry_after if e.retry_after is not None else 2**attempt)

Failures without a response, such as connection errors, have a status code of 0. Invalid arguments, such as an empty UUID, a page below 1, a search query with fewer than 2 non-whitespace characters, or an access token that is empty or contains whitespace, are rejected by the client immediately with a ValueError.

Testing

The unit tests run without a server:

pip install -e . --group dev
pytest

The integration tests run against a live Federation server, and the repository includes a docker-compose.yml that brings up the reference FederationLib server, MariaDB, and Redis on http://localhost:7000:

docker compose up -d
pytest -m integration

By default the tests expect the server at http://localhost:7000 with access token abcdefghijklmnopqrstuvwxyz123456 (matching the compose defaults). To test against a different server, set the SERVER_ENDPOINT and SERVER_ACCESS_TOKEN environment variables:

SERVER_ENDPOINT=http://localhost:7000 SERVER_ACCESS_TOKEN=my-token pytest -m integration

The synchronous client, client.py, is generated from the asynchronous client, async_client.py, run python scripts/unasync.py after changing it.

License

This project is licensed under the MIT License - see the LICENSE file for details.