Skip to content

cattle_grid.account

cattle_grid.account.account

account_for_actor async 

account_for_actor(
    session: AsyncSession, actor: str
) -> Account | None

Given an actor_id returns the corresponding account or None

Source code in cattle_grid/account/account.py 
142
143
144
145
146
147
148
149
150
151
async def account_for_actor(session: AsyncSession, actor: str) -> Account | None:
    """Given an actor_id returns the corresponding account or None"""
    actor_for_account = await session.scalar(
        select(ActorForAccount).where(ActorForAccount.actor == actor)
    )

    if not actor_for_account:
        return None

    return actor_for_account.account

account_with_name_password async 

account_with_name_password(
    session: AsyncSession, name: str, password: str
) -> Account | None

Retrieves account for given name and password

Source code in cattle_grid/account/account.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def account_with_name_password(
    session: AsyncSession, name: str, password: str
) -> Account | None:
    """Retrieves account for given name and password"""
    account = await session.scalar(select(Account).where(Account.name == name))
    if account is None:
        return None

    try:
        password_hasher.verify(account.password_hash, password)
    except argon2.exceptions.VerifyMismatchError:
        logger.warning("Got wrong password for %s", name)
        return None

    # Implement rehash?
    # https://argon2-cffi.readthedocs.io/en/stable/howto.html

    return account

actor_for_actor_id async 

actor_for_actor_id(
    session: AsyncSession, actor_id: str
) -> ActorForAccount | None

Returns the ActorForAccount for the given actor_id

Source code in cattle_grid/account/account.py 
178
179
180
181
182
183
184
async def actor_for_actor_id(
    session: AsyncSession, actor_id: str
) -> ActorForAccount | None:
    """Returns the ActorForAccount for the given actor_id"""
    return await session.scalar(
        select(ActorForAccount).where(ActorForAccount.actor == actor_id)
    )

add_actor_to_account async 

add_actor_to_account(
    session: AsyncSession,
    account: Account,
    actor: Actor,
    name: str = "added",
) -> None

Adds the actor to the account

Source code in cattle_grid/account/account.py 
154
155
156
157
158
159
async def add_actor_to_account(
    session: AsyncSession, account: Account, actor: Actor, name: str = "added"
) -> None:
    """Adds the actor to the account"""
    session.add(ActorForAccount(account=account, actor=actor.actor_id, name=name))
    await session.commit()

add_actor_to_group async 

add_actor_to_group(
    session: AsyncSession,
    actor: ActorForAccount,
    group_name: str,
) -> None

Adds the actor to the group

Source code in cattle_grid/account/account.py 
162
163
164
165
166
167
async def add_actor_to_group(
    session: AsyncSession, actor: ActorForAccount, group_name: str
) -> None:
    """Adds the actor to the group"""
    session.add(ActorGroup(actor=actor, name=group_name))
    await session.commit()

add_permission async 

add_permission(
    session: AsyncSession, account: Account, permission: str
) -> None

Adds permission to account

Source code in cattle_grid/account/account.py 
118
119
120
121
122
123
async def add_permission(
    session: AsyncSession, account: Account, permission: str
) -> None:
    """Adds permission to account"""
    session.add(Permission(account=account, name=permission))
    await session.commit()

create_account async 

create_account(
    session: AsyncSession,
    name: str,
    password: str,
    settings=get_settings(),
    permissions: list[str] = [],
    meta_information: dict[str, str] = {},
) -> Account | None

Creates a new account for name and password

Source code in cattle_grid/account/account.py 
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
async def create_account(
    session: AsyncSession,
    name: str,
    password: str,
    settings=get_settings(),
    permissions: list[str] = [],
    meta_information: dict[str, str] = {},
) -> Account | None:
    """Creates a new account for name and password"""
    if await session.scalar(select(Account).where(Account.name == name)):
        raise AccountAlreadyExists("Account already exists")

    if not re.match(settings.account.allowed_name_regex, name):  # type: ignore
        raise InvalidAccountName("Account name does not match allowed format")

    if name in settings.account.forbidden_names:  # type: ignore
        raise InvalidAccountName("Account name is forbidden")

    account = Account(
        name=name,
        password_hash=password_hasher.hash(password),
        meta_information=meta_information,
    )
    session.add(account)
    await session.commit()

    for permission in permissions:
        await add_permission(session, account, permission)
    return account

delete_account async 

delete_account(
    session: AsyncSession,
    name: str,
    password: str | None = None,
    force: bool = False,
) -> None

Deletes account for given account name and password

If password is wrong or account does not exist, raises a WrongPassword exception

Source code in cattle_grid/account/account.py 
 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
async def delete_account(
    session: AsyncSession, name: str, password: str | None = None, force: bool = False
) -> None:
    """Deletes account for given account name and password

    If password is wrong or account does not exist,
    raises a WrongPassword exception"""
    if force:
        account = await session.scalar(select(Account).where(Account.name == name))
    else:
        if not password:
            raise ValueError("Password must be provided if not forcing deletion")
        account = await account_with_name_password(session, name, password)

    if account is None:
        raise WrongPassword(
            "Either the account does not exist or the password is wrong"
        )

    await session.refresh(account, attribute_names=["actors"])

    active_actors = [a for a in account.actors if a.status == ActorStatus.active]
    if len(active_actors) > 0:
        logger.warning(
            "Deleting account with active actors: %s", [a.actor for a in active_actors]
        )

    await session.delete(account)
    await session.commit()

group_names_for_actor async 

group_names_for_actor(
    session: AsyncSession, actor: ActorForAccount
) -> list[str]

Returns the group names for an actor

Source code in cattle_grid/account/account.py 
170
171
172
173
174
175
async def group_names_for_actor(
    session: AsyncSession, actor: ActorForAccount
) -> list[str]:
    """Returns the group names for an actor"""
    await session.refresh(actor, attribute_names=["groups"])
    return [g.name for g in actor.groups]

list_permissions 

list_permissions(account: Account) -> list[str]

Returns list of permissions for account

Source code in cattle_grid/account/account.py 
137
138
139
def list_permissions(account: Account) -> list[str]:
    """Returns list of permissions for account"""
    return [p.name for p in account.permissions]

remove_permission async 

remove_permission(
    session: AsyncSession, account: Account, permission: str
) -> None

Removes permission from account

Source code in cattle_grid/account/account.py 
126
127
128
129
130
131
132
133
134
async def remove_permission(
    session: AsyncSession, account: Account, permission: str
) -> None:
    """Removes permission from account"""
    await session.execute(
        delete(Permission)
        .where(Permission.account == account)
        .where(Permission.name == permission)
    )

cattle_grid.account.permissions

allowed_base_urls async 

allowed_base_urls(
    session: AsyncSession, account: Account
) -> list[str]

Returns the set of base_urls the account is allowed to use to create an actor

Source code in cattle_grid/account/permissions.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
async def allowed_base_urls(session: AsyncSession, account: Account) -> list[str]:
    """Returns the set of base_urls the account
    is allowed to use to create an actor"""
    settings = global_container.config
    await session.refresh(account, attribute_names=["permissions"])

    permissions = list_permissions(account)

    if "admin" in permissions:
        frontend_settings = settings.get("frontend", {})  # type:ignore

        return frontend_settings.get("base_urls", [])

    permission_settings = settings.get("permissions", {})  # type:ignore

    return sum(
        (permission_settings.get(p, {}).get("base_urls", []) for p in permissions),
        [],
    )

can_create_actor_at_base_url async 

can_create_actor_at_base_url(
    session: AsyncSession, account: Account, base_url
) -> bool

Checks if the account is allowed to create an actor at the base url

Source code in cattle_grid/account/permissions.py 
30
31
32
33
34
35
36
37
async def can_create_actor_at_base_url(
    session: AsyncSession, account: Account, base_url
) -> bool:
    """Checks if the account is allowed to create an actor
    at the base url"""
    allowed_urls = await allowed_base_urls(session, account)

    return base_url in allowed_urls