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 OFD-compliant Federation server
• Content scanning with entity resolution, classification, and risk assessment
• 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}")

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()

Content scanning

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

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 be at least 2 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

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.

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.

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:

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.

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), 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}")

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 shorter than 2 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.