Skip to content

cattle_grid.config

cattle_grid.config

Implements loading the configuration

auth

AuthConfig

Bases: BaseModel

Configures the Authorization layer

Parameters:

Name Type Description Default
actor_id str
required
actor_acct_id str
required
public_key str
required
private_key str
required
domain_blocks Set[str]
required
require_signature_for_activity_pub bool
True
Source code in cattle_grid/config/auth.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class AuthConfig(BaseModel):
    """Configures the Authorization layer"""

    actor_id: str
    """actor_id for the Application actor used to fetch public keys"""
    actor_acct_id: str
    """acct uri of the Application Actor used to fetch public keys"""
    public_key: str
    """Public key of the Application actor"""
    private_key: str
    """Private key of the Application actor"""

    domain_blocks: Set[str]
    """Set of blocked domains"""

    require_signature_for_activity_pub: bool = True
    """If set to true, all requests with accept type that match activitypub must be signed"""

    @field_serializer("domain_blocks")
    def serialize_domain_blocks(self, domain_blocks: Set[str], _info):
        return list(domain_blocks)
actor_acct_id instance-attribute

acct uri of the Application Actor used to fetch public keys

actor_id instance-attribute

actor_id for the Application actor used to fetch public keys

domain_blocks instance-attribute

Set of blocked domains

private_key instance-attribute

Private key of the Application actor

public_key instance-attribute

Public key of the Application actor

require_signature_for_activity_pub = True class-attribute instance-attribute

If set to true, all requests with accept type that match activitypub must be signed

get_auth_config(settings=get_settings()) cached

Returns the configuration for authorization

Returns:

Type Description
AuthConfig
Source code in cattle_grid/config/auth.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@lru_cache
def get_auth_config(settings: Dynaconf = get_settings()) -> AuthConfig:
    """Returns the configuration for authorization

    :returns:
    """
    try:
        auth = settings.get("auth")
        if not auth:
            raise AuthNotConfigured("No authorization configuration found")

        data = {
            "actor_id": auth.actor_id,
            "actor_acct_id": auth.actor_acct_id,
            "public_key": auth.public_key,
            "private_key": auth.private_key,
            "domain_blocks": auth.get("domain_blocks", []),
            "require_signature_for_activity_pub": auth.get(
                "require_signature_for_activity_pub", True
            ),
        }
        return AuthConfig.model_validate(data)

    except ValidationError:
        raise AuthNotConfigured(
            "Authorization not configured or configured incorrectly"
        )

new_auth_config(actor_id, username=None)

Creates a new authorization configuration

Source code in cattle_grid/config/auth.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def new_auth_config(actor_id: str, username: str | None = None) -> AuthConfig:
    """Creates a new authorization configuration"""
    if not username:
        username = secrets.token_urlsafe(12)

    domain = urlparse(actor_id).netloc
    acct_uri = f"acct:{username}@{domain}"

    public_key, private_key = generate_rsa_public_private_key()

    auth_config = AuthConfig(
        actor_id=actor_id,
        actor_acct_id=acct_uri,
        public_key=public_key,
        private_key=private_key,
        domain_blocks=set(),
    )

    return auth_config

save_auth_config(filename, config)

Saves the authorization configuration to a file

Source code in cattle_grid/config/auth.py 
94
95
96
97
def save_auth_config(filename: str, config: AuthConfig) -> None:
    """Saves the authorization configuration to a file"""
    with open(filename, "wb") as fp:
        tomli_w.dump({"auth": config.model_dump()}, fp, multiline_strings=True)

messaging

account_exchange(settings=get_settings()) cached

Returns the exchange used to distribute messages between accounts

Returns:

Type Description
RabbitExchange
Source code in cattle_grid/config/messaging.py 
29
30
31
32
33
34
35
36
37
38
39
@lru_cache
def account_exchange(settings=get_settings()) -> RabbitExchange:
    """Returns the exchange used to distribute messages between accounts

    :returns:
    """
    durable = True if settings.activity_pub.account_exchange == "amq.topic" else False

    return RabbitExchange(
        settings.activity_pub.account_exchange, type=ExchangeType.TOPIC, durable=durable
    )

broker(settings=get_settings())

Returns the rabbitmq broker

Returns:

Type Description
RabbitBroker
Source code in cattle_grid/config/messaging.py 
45
46
47
48
49
50
51
52
53
54
55
def broker(settings=get_settings()) -> RabbitBroker:
    """Returns the rabbitmq broker

    :returns:
    """
    global current_broker

    if current_broker is None:
        current_broker = RabbitBroker(settings.amqp_uri)

    return current_broker

exchange(settings=get_settings()) cached

Returns the public exchange used to process

Returns:

Type Description
RabbitExchange
Source code in cattle_grid/config/messaging.py 
20
21
22
23
24
25
26
@lru_cache
def exchange(settings=get_settings()) -> RabbitExchange:
    """Returns the public exchange used to process

    :returns:
    """
    return RabbitExchange(settings.activity_pub.exchange, type=ExchangeType.TOPIC)

internal_exchange(settings=get_settings()) cached

The internal exchange used to process ActivityPub messages related to the social graph

Returns:

Type Description
RabbitExchange
Source code in cattle_grid/config/messaging.py 
 8
 9
10
11
12
13
14
15
16
17
@lru_cache
def internal_exchange(settings=get_settings()) -> RabbitExchange:
    """The internal exchange used to process
    ActivityPub messages related to the social graph

    :returns:
    """
    return RabbitExchange(
        settings.activity_pub.internal_exchange, type=ExchangeType.TOPIC
    )

validators

account_validations = [Validator('account.forbidden_names', default=lambda a, b: list(['bovine', 'cattle_grid', 'admin', 'guest']), cast=list), Validator('account.allowed_name_regex', cast=str, default='^[a-zA-Z0-9_]{1,16}$')] module-attribute

Validators for the account

activity_pub_validators = [Validator('activity_pub.internal_exchange', default='cattle_grid_internal'), Validator('activity_pub.exchange', default='cattle_grid'), Validator('activity_pub.account_exchange', default='amq.topic')] module-attribute

Validators for ActivityPub

auth_validators = [Validator('auth.require_signature_for_activity_pub', default=True)] module-attribute

Validates the authentication configuration

base_validators = [Validator('amqp_uri', default='amqp://localhost'), Validator('db_uri', default='sqlite://cattle_grid.db'), Validator('enable_reporting', cast=bool, default=False)] module-attribute

Validates the basic configuration

extensions_validations = [Validator('extensions', default=lambda a, b: list([]), cast=list)] module-attribute

Validators for the plugins

frontend_validations = [Validator('frontend.base_urls', default=lambda a, b: list([]), cast=lambda x: [str(y) for y in x], condition=lambda items: all(x.startswith('http://') or x.startswith('https://') for x in items))] module-attribute

Validators for the frontend

gateway_admin_validations = [Validator('gateway.admin.enable', cast=bool, default=False), Validator('gateway.admin.enable_reset', cast=bool, default=False)] module-attribute

Validators for the gateway

plugins_validations = [Validator('plugins', default=lambda a, b: list([]), cast=list)] module-attribute

Validators for the plugins