Skip to content

.auth

cattle_grid.auth

create_app(filenames)

Allows running just the auth endpoint

Source code in cattle_grid/auth/__init__.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def create_app(filenames):
    """Allows running just the auth endpoint"""

    config = load_settings(filenames)

    @asynccontextmanager
    async def lifespan(app):
        async with database(config):
            await tortoise.Tortoise.generate_schemas()
            yield

    app = FastAPI(
        lifespan=lifespan,
        title="cattle_grid.auth",
        description="""Authorization server for Fediverse applications. It basically checks HTTP Signatures for you.""",
        version=__version__,
    )

    app.include_router(create_auth_router(get_auth_config(config), tags=[]))

    return app

__main__

new_config(actor_id, config_file, username)

Creates a new authorization configuration, including generating a public and private key.

Argument is the actor_id for the fetch actor, e.g. http://cattle_grid/fetch_actor

Source code in cattle_grid/auth/__main__.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@auth_cli.command()
@click.argument("actor_id", required=True)
@click.option(
    "--config_file",
    help="Filename to save the configuration to",
    default="cattle_grid_auth.toml",
)
@click.option(
    "--username",
    help="Specify a username for the fetch actor. Otherwise it is automatically generated",
)
def new_config(actor_id, config_file, username):
    """Creates a new authorization configuration,
    including generating a public and private key.

    Argument is the actor_id for the fetch actor,
    e.g. http://cattle_grid/fetch_actor
    """
    config = new_auth_config(actor_id, username=username)

    save_auth_config(config_file, config)

http_util

AcceptEntry dataclass

AcceptEntry(content_type: str, profile: str | None = None, quality: float = 1.0)

Parameters:

Name Type Description Default
content_type str
required
profile str | None
None
quality float
1.0
Source code in cattle_grid/auth/http_util.py 
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
@dataclass
class AcceptEntry:
    content_type: str
    profile: str | None = None
    quality: float = 1.0

    @staticmethod
    def from_header(piece):
        return AcceptEntry(
            content_type=str(piece[0]),
            profile=piece[1].get("profile", None),
            quality=float(piece[1].get("q", 1.0)),
        )

    def to_content_type(self) -> ContentType:
        if self.content_type == "text/html":
            return ContentType.html
        elif self.content_type == "application/activity+json":
            return ContentType.activity_pub
        elif (
            self.content_type == "application/ld+json"
            and self.profile == "https://www.w3.org/ns/activitystreams"
        ):
            return ContentType.activity_pub
        else:
            return ContentType.other

ContentType

Bases: StrEnum

The content type of the response

Source code in cattle_grid/auth/http_util.py 
 7
 8
 9
10
11
12
class ContentType(StrEnum):
    """The content type of the response"""

    activity_pub = auto()
    html = auto()
    other = auto()

parse_accept_header(header) 

>>> header = 'application/activity+json, application/ld+json; profile="https://www.w3.org/ns/activitystreams", text/html;q=0.1'
>>> parse_accept_header(header)
[AcceptEntry(content_type='application/activity+json', profile=None, quality=1.0),
    AcceptEntry(content_type='application/ld+json', profile='https://www.w3.org/ns/activitystreams', quality=1.0),
    AcceptEntry(content_type='text/html', profile=None, quality=0.1)]
Source code in cattle_grid/auth/http_util.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def parse_accept_header(header):
    """
    ```pycon
    >>> header = 'application/activity+json, application/ld+json; profile="https://www.w3.org/ns/activitystreams", text/html;q=0.1'
    >>> parse_accept_header(header)
    [AcceptEntry(content_type='application/activity+json', profile=None, quality=1.0),
        AcceptEntry(content_type='application/ld+json', profile='https://www.w3.org/ns/activitystreams', quality=1.0),
        AcceptEntry(content_type='text/html', profile=None, quality=0.1)]

    ```
    """

    parsed = http_sf.parse(header.encode(), tltype="list")

    return [AcceptEntry.from_header(piece) for piece in parsed]

model

RemoteIdentity

Bases: Model

Represents the information about a remote identity

Source code in cattle_grid/auth/model.py 
 5
 6
 7
 8
 9
10
11
class RemoteIdentity(Model):
    """Represents the information about a remote identity"""

    id = fields.IntField(primary_key=True)
    key_id = fields.CharField(max_length=512, unique=True)
    controller = fields.CharField(max_length=512)
    public_key = fields.CharField(max_length=1024)

public_key_cache

PublicKeyCache dataclass

Caches public keys in the database and fetches them using bovine_actor

Parameters:

Name Type Description Default
bovine_actor BovineActor
required
Source code in cattle_grid/auth/public_key_cache.py 
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
67
68
69
70
71
72
73
74
75
@dataclass
class PublicKeyCache:
    """Caches public keys in the database and fetches them
    using bovine_actor"""

    bovine_actor: BovineActor
    """used to fetch the public key"""

    async def cryptographic_identifier(
        self, key_id: str
    ) -> CryptographicIdentifier | Literal["gone"] | None:
        """Returns "gone" if Tombstone

        :param key_id: URI of the public key to fetch
        :returns:
        """

        try:
            result = await self.bovine_actor.get(key_id)
            if result.get("type") == "Tombstone":
                logger.info("Got Tombstone for %s", key_id)
                return "gone"

            public_key, owner = actor_object_to_public_key(result, key_id)
            return CryptographicIdentifier.from_pem(public_key, owner)
        except Exception as e:
            logger.info("Failed to fetch public key for %s with %s", key_id, repr(e))
            logger.exception(e)
            return None

    async def from_cache(self, key_id: str) -> CryptographicIdentifier | None:
        identity = await RemoteIdentity.get_or_none(key_id=key_id)

        if identity is None:
            identifier = await self.cryptographic_identifier(key_id)
            if identifier is None:
                return None

            if identifier == "gone":
                await RemoteIdentity.create(
                    key_id=key_id, public_key="gone", controller="gone"
                )
                return None

            try:
                controller, public_key = identifier.as_tuple()
                identity = await RemoteIdentity.create(
                    key_id=key_id, public_key=public_key, controller=controller
                )
                await identity.save()
            except Exception as e:
                logger.exception(e)
            return identifier

        if identity.controller == "gone":
            return None

        return CryptographicIdentifier.from_tuple(
            identity.controller, identity.public_key
        )
bovine_actor instance-attribute

used to fetch the public key

cryptographic_identifier(key_id) async

Returns “gone” if Tombstone

Parameters:

Name Type Description Default
key_id str

URI of the public key to fetch

 required

Returns:

Type Description
CryptographicIdentifier | Literal['gone'] | None
Source code in cattle_grid/auth/public_key_cache.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
async def cryptographic_identifier(
    self, key_id: str
) -> CryptographicIdentifier | Literal["gone"] | None:
    """Returns "gone" if Tombstone

    :param key_id: URI of the public key to fetch
    :returns:
    """

    try:
        result = await self.bovine_actor.get(key_id)
        if result.get("type") == "Tombstone":
            logger.info("Got Tombstone for %s", key_id)
            return "gone"

        public_key, owner = actor_object_to_public_key(result, key_id)
        return CryptographicIdentifier.from_pem(public_key, owner)
    except Exception as e:
        logger.info("Failed to fetch public key for %s with %s", key_id, repr(e))
        logger.exception(e)
        return None

router

ReverseProxyHeaders

Bases: BaseModel

Headers set by the reverse proxy

Parameters:

Name Type Description Default
x_original_method str
'get'
x_original_uri str | None
None
x_original_host str | None
None
x_forwarded_proto str
'http'
Source code in cattle_grid/auth/router.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class ReverseProxyHeaders(BaseModel):
    """Headers set by the reverse proxy"""

    x_original_method: str = "get"
    """The original used method"""

    x_original_uri: str | None = None
    """The original request uri"""

    x_original_host: str | None = None
    """The original used host"""

    x_forwarded_proto: str = "http"
    """The protocol being used"""
x_forwarded_proto = 'http' class-attribute instance-attribute

The protocol being used

x_original_host = None class-attribute instance-attribute

The original used host

x_original_method = 'get' class-attribute instance-attribute

The original used method

x_original_uri = None class-attribute instance-attribute

The original request uri

create_auth_router(config, tags=['auth'])

Adds the authorization endpoint to the app

Source code in cattle_grid/auth/router.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def create_auth_router(
    config: AuthConfig, tags: List[str | Enum] = ["auth"]
) -> APIRouter:
    """Adds the authorization endpoint to the app"""

    bovine_actor = config_to_bovine_actor(config)
    public_key_cache = PublicKeyCache(bovine_actor)
    signature_checker = SignatureChecker(
        public_key_cache.from_cache, skip_digest_check=True
    )

    actor_path = str(urlparse(config.actor_id).path)
    webfinger_jrd = webfinger_response(config.actor_acct_id, config.actor_id)

    username, _ = parse_fediverse_handle(config.actor_acct_id.removeprefix("acct:"))
    actor_object = Actor(
        id=config.actor_id,
        type="Service",
        public_key=config.public_key,
        preferred_username=username,
        public_key_name="mykey",
    ).build()

    @asynccontextmanager
    async def lifespan(app):
        async with aiohttp.ClientSession() as session:
            await bovine_actor.init(session=session)
            yield

    router = APIRouter(lifespan=lifespan, tags=tags)

    @router.get(
        actor_path,
        response_class=ActivityResponse,
    )
    async def handle_get_actor():
        """Returns the actor profile of the
        fetch actor used to retrieve public keys, e.g.

        ```json
        {
            "type": "Service",
            "id": "https://your-domain.example/cattle_grid_actor",
            ...
        }
        ```
        """
        return actor_object

    @router.get("/.well-known/webfinger")
    async def webfinger(resource: str) -> JrdData:
        """If resource is the actor corresponding to the actor fetching
        public keys, returns the corresponding Jrd. Otherwise returns
        not found"""
        if resource != config.actor_acct_id:
            raise HTTPException(404)
        return webfinger_jrd

    @router.get(
        "/auth",
        responses={
            401: {"description": "The signature was invalid"},
            403: {"description": "Request was blocked"},
        },
    )
    async def verify_signature(
        request: Request,
        response: Response,
        reverse_proxy_headers: Annotated[ReverseProxyHeaders, Header()],
    ):
        """Takes the request and checks signature. If signature check
        fails a 401 is returned. If the domain the public key belongs
        to is blocked, a 403 is returned.

        If the request is valid. The controller corresponding to
        the signature is set in the response header `X-CATTLE-GRID-REQUESTER`.

        Note: More headers than the ones listed below can be used
        to verify a signature.
        """
        headers = MutableHeaders(request.headers)

        servable_content_types = should_serve(headers.get("accept"))

        if "signature" not in headers:
            if ContentType.html in servable_content_types:
                response.headers["x-cattle-grid-should-serve"] = "html"
                return ""
            elif ContentType.other in servable_content_types:
                response.headers["x-cattle-grid-should-serve"] = "other"
                return ""
            elif config.require_signature_for_activity_pub:
                raise HTTPException(401)
            else:
                return ""

        if reverse_proxy_headers.x_original_host:
            headers["Host"] = reverse_proxy_headers.x_original_host

        url = f"{reverse_proxy_headers.x_forwarded_proto}://{reverse_proxy_headers.x_original_host}{reverse_proxy_headers.x_original_uri}"

        controller = await signature_checker.validate_signature(
            reverse_proxy_headers.x_original_method.lower(),
            url,
            dict(headers.items()),
            None,
        )

        if controller:
            if check_block(config.domain_blocks, controller):
                logger.info("Blocked a request by %s", controller)
                raise HTTPException(403)
            response.headers["x-cattle-grid-requester"] = controller

            return ""

        logger.info(
            "invalid signature for request to %s => access denied",
            request.headers.get("X-Original-Uri", ""),
        )

        raise HTTPException(401)

    return router

util

check_block(domain_blocks, controller)

Checks if a controller’s domain is in block list

>>> check_block({"blocked.example"}, "http://actor.example/path")
False

>>> check_block({"blocked.example"}, "http://blocked.example/path")
True
Source code in cattle_grid/auth/util.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def check_block(domain_blocks: Set[str], controller: str) -> bool:
    """Checks if a controller's domain is in block list

    ```pycon
    >>> check_block({"blocked.example"}, "http://actor.example/path")
    False

    >>> check_block({"blocked.example"}, "http://blocked.example/path")
    True

    ```
    """
    try:
        domain = urlparse(controller).netloc
        return domain in domain_blocks
    except Exception as e:
        logger.warning("Something went wrong with %s", repr(e))
        return True