Skip to content

cattle_grid.account.rabbit

cattle_grid.account.rabbit

Implementation of a HTTP auth backend for rabbitmq.

A possible configuration of RabbitMQ is

/etc/rabbitmq/conf.d/03_http_auth.conf
auth_backends.1 = internal

auth_backends.2 = http
auth_http.http_method = post
auth_http.user_path = http://cattle_grid_app/rabbitmq/user
auth_http.vhost_path = http://cattle_grid_app/rabbitmq/vhost
auth_http.resource_path = http://cattle_grid_app/rabbitmq/resource
auth_http.topic_path = http://cattle_grid_app/rabbitmq/topic

Here, we use auth_backend = internal for the user corresponding to the cattle_grid processes. As cattle_grid connects to RabbitMQ on startup, it cannot authenticate itself.

resource_auth async 

resource_auth() -> str

Always allowed

Source code in cattle_grid/account/rabbit.py 
59
60
61
62
@rabbit_router.post("/resource", response_class=PlainTextResponse)
async def resource_auth() -> str:
    """Always allowed"""
    return "allow"

topic_auth async 

topic_auth(
    username: Annotated[str, Form()],
    name: Annotated[str, Form()],
    routing_key: Annotated[str, Form()],
) -> str

Checks if topic is allowed. Currently allowed are

exchange = "amq.topic"

and the routing keys send.username and receive.username

Source code in cattle_grid/account/rabbit.py 
 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
@rabbit_router.post(
    "/topic",
    response_class=PlainTextResponse,
)
async def topic_auth(
    username: Annotated[str, Form()],
    name: Annotated[str, Form()],
    routing_key: Annotated[str, Form()],
) -> str:
    """Checks if topic is allowed. Currently allowed are

    ```
    exchange = "amq.topic"
    ```

    and the routing keys `send.username` and `receive.username`
    """
    if name != "amq.topic":
        logger.warning("User %s tried to access exchange %s", username, name)
        return "deny"
    if not validate_routing_key(username, routing_key):
        logger.warning(
            "User %s tried to subscribe to routing_key %s",
            username,
            routing_key,
        )
        return "deny"

    return "allow"

user_auth async 

user_auth(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()],
    session: SqlSession,
) -> str

Checks login with username/password

Source code in cattle_grid/account/rabbit.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@rabbit_router.post("/user", response_class=PlainTextResponse)
async def user_auth(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()],
    session: SqlSession,
) -> str:
    """Checks login with username/password"""
    user = await account_with_name_password(session, username, password)

    if not user:
        logger.warning("Failed login to message broker with username '%s'", username)
        return "deny"

    return "allow"

validate_routing_key 

validate_routing_key(username: str, routing_key: str)

Rules for the routing key, e.g.

>>> validate_routing_key("alice", "send.alice.trigger")
True

>>> validate_routing_key("alice", "send.bob.trigger")
False
Source code in cattle_grid/account/rabbit.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def validate_routing_key(username: str, routing_key: str):
    """Rules for the routing key, e.g.

    ```pycon
    >>> validate_routing_key("alice", "send.alice.trigger")
    True

    >>> validate_routing_key("alice", "send.bob.trigger")
    False

    ```
    """
    if routing_key.startswith(f"send.{username}."):
        return True
    if routing_key.startswith(f"receive.{username}."):
        return True
    if routing_key == f"error.{username}":
        return True
    return False

vhost_auth async 

vhost_auth(
    username: Annotated[str, Form()],
    vhost: Annotated[str, Form()],
) -> str

Authentication for vhosts, currently only “/” is allowed

Source code in cattle_grid/account/rabbit.py 
48
49
50
51
52
53
54
55
56
@rabbit_router.post("/vhost", response_class=PlainTextResponse)
async def vhost_auth(
    username: Annotated[str, Form()], vhost: Annotated[str, Form()]
) -> str:
    """Authentication for vhosts, currently only "/" is allowed"""
    if vhost != "/":
        logger.warning("User %s tried to access vhost %s", username, vhost)
        return "deny"
    return "allow"