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

actor_id for the Application actor used to fetch public keys

 required
actor_acct_id str

acct uri of the Application Actor used to fetch public keys

 required
public_key str

Public key of the Application actor

 required
private_key str

Private key of the Application actor

 required
domain_blocks Set[str]

Set of blocked domains

 required
require_signature_for_activity_pub bool

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

 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 
actor_acct_id: str

acct uri of the Application Actor used to fetch public keys

actor_id instance-attribute 
actor_id: str

actor_id for the Application actor used to fetch public keys

domain_blocks instance-attribute 
domain_blocks: Set[str]

Set of blocked domains

private_key instance-attribute 
private_key: str

Private key of the Application actor

public_key instance-attribute 
public_key: str

Public key of the Application actor

require_signature_for_activity_pub class-attribute instance-attribute 
require_signature_for_activity_pub: bool = True

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

get_auth_config cached 

get_auth_config(
    settings: Dynaconf = get_settings(),
) -> AuthConfig

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")  # type:ignore
        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 

new_auth_config(
    actor_id: str, username: str | None = None
) -> AuthConfig

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 

save_auth_config(filename: str, config: AuthConfig) -> None

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)

logging

configure_logging 

configure_logging(settings)

Sets the log level according to the settings

Source code in cattle_grid/config/logging.py 
4
5
6
7
8
9
def configure_logging(settings):
    """Sets the log level according to the settings"""
    log_config = settings.get("logging", {})

    for name, level in log_config.items():
        logging.getLogger(name).setLevel(level.upper())

rewrite

RewriteConfiguration dataclass

Configuration of the rewrite process

Parameters:

Name Type Description Default
ruleset dict[str, RewriteRules]

mapping between group names and rewrite rules

 required
Source code in cattle_grid/config/rewrite.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@dataclass
class RewriteConfiguration:
    """Configuration of the rewrite process"""

    ruleset: dict[str, RewriteRules] = field(
        metadata={"description": "mapping between group names and rewrite rules"}
    )

    def rewrite(self, method_name: str, group_names: list[str]) -> str:
        """rewrites the method_name"""
        for group_name in group_names:
            rules = self.ruleset.get(group_name)

            if rules and method_name in rules.rules:
                return rules.rules[method_name]

        return method_name

    def add_rules(
        self, group_name: str, rules: dict[str, str], overwrite: bool = False
    ):
        """Adds new rules to the ruleset"""
        if not overwrite and group_name in self.ruleset:
            return

        self.ruleset[group_name] = RewriteRules.model_validate({"rules": rules})

    @staticmethod
    def from_rules(rules: dict[str, dict[str, str]] | None):
        """
        Constructs the RewriteConfiguration

        ```
        >>> rules = {"old": "new"}
        >>> config = RewriteConfiguration.from_rules({"group": rules})
        >>> config
        RewriteConfiguration(ruleset={'group': RewriteRules(rules={'old': 'new'})})

        >>> group_names = ["group"]
        >>> config.rewrite("unchanged", group_names)
        'unchanged'

        >>> config.rewrite("old", group_names)
        'new'

        ```
        """
        if rules:
            mapped = {
                group: RewriteRules.model_validate({"rules": rule})
                for group, rule in rules.items()
            }
        else:
            mapped = {}

        return RewriteConfiguration(ruleset=mapped)
add_rules 
add_rules(
    group_name: str,
    rules: dict[str, str],
    overwrite: bool = False,
)

Adds new rules to the ruleset

Source code in cattle_grid/config/rewrite.py 
29
30
31
32
33
34
35
36
def add_rules(
    self, group_name: str, rules: dict[str, str], overwrite: bool = False
):
    """Adds new rules to the ruleset"""
    if not overwrite and group_name in self.ruleset:
        return

    self.ruleset[group_name] = RewriteRules.model_validate({"rules": rules})
from_rules staticmethod 
from_rules(rules: dict[str, dict[str, str]] | None)

Constructs the RewriteConfiguration

>>> rules = {"old": "new"}
>>> config = RewriteConfiguration.from_rules({"group": rules})
>>> config
RewriteConfiguration(ruleset={'group': RewriteRules(rules={'old': 'new'})})

>>> group_names = ["group"]
>>> config.rewrite("unchanged", group_names)
'unchanged'

>>> config.rewrite("old", group_names)
'new'
Source code in cattle_grid/config/rewrite.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@staticmethod
def from_rules(rules: dict[str, dict[str, str]] | None):
    """
    Constructs the RewriteConfiguration

    ```
    >>> rules = {"old": "new"}
    >>> config = RewriteConfiguration.from_rules({"group": rules})
    >>> config
    RewriteConfiguration(ruleset={'group': RewriteRules(rules={'old': 'new'})})

    >>> group_names = ["group"]
    >>> config.rewrite("unchanged", group_names)
    'unchanged'

    >>> config.rewrite("old", group_names)
    'new'

    ```
    """
    if rules:
        mapped = {
            group: RewriteRules.model_validate({"rules": rule})
            for group, rule in rules.items()
        }
    else:
        mapped = {}

    return RewriteConfiguration(ruleset=mapped)
rewrite 
rewrite(method_name: str, group_names: list[str]) -> str

rewrites the method_name

Source code in cattle_grid/config/rewrite.py 
19
20
21
22
23
24
25
26
27
def rewrite(self, method_name: str, group_names: list[str]) -> str:
    """rewrites the method_name"""
    for group_name in group_names:
        rules = self.ruleset.get(group_name)

        if rules and method_name in rules.rules:
            return rules.rules[method_name]

    return method_name

RewriteRules

Bases: BaseModel

Describes the rewrite rules

Parameters:

Name Type Description Default
rules dict[str, str]

Mapping from old rule to new rule

 required
Source code in cattle_grid/config/rewrite.py 
5
6
7
8
class RewriteRules(BaseModel):
    """Describes the rewrite rules"""

    rules: dict[str, str] = Field(description="Mapping from old rule to new rule")

validators

account_validations module-attribute 

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

Validators for the account

activity_pub_validators module-attribute 

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

Validators for ActivityPub

auth_validators module-attribute 

auth_validators = [
    Validator(
        "auth.require_signature_for_activity_pub",
        default=True,
    )
]

Validates the authentication configuration

base_validators module-attribute 

base_validators = [
    Validator("amqp_uri", default="amqp://localhost"),
    Validator("db_uri", default="sqlite:///cattle_grid.db"),
    Validator("enable_reporting", cast=bool, default=False),
    Validator("processor_in_app", cast=bool, default=False),
    Validator("permissions", default={}),
]

Validates the basic configuration

extensions_validations module-attribute 

extensions_validations = [
    Validator(
        "extensions",
        default=lambda a, b: list([]),
        cast=list,
    )
]

Validators for the plugins

frontend_validations module-attribute 

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

Validators for the frontend

gateway_admin_validations module-attribute 

gateway_admin_validations = [
    Validator(
        "gateway.admin.enable", cast=bool, default=False
    ),
    Validator(
        "gateway.admin.enable_reset",
        cast=bool,
        default=False,
    ),
]

Validators for the gateway

plugins_validations module-attribute 

plugins_validations = [
    Validator(
        "plugins", default=lambda a, b: list([]), cast=list
    )
]

Validators for the plugins

testing_validators module-attribute 

testing_validators = [
    Validator("testing.enable", cast=bool, default=False),
    Validator(
        "testing.accounts",
        default=lambda a, b: list([]),
        cast=list,
    ),
]

Validators for testing