34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
"""RBAC-Durchsetzung: Rolle × Hostgruppe (siehe Konzept 4.6)."""
|
||
from __future__ import annotations
|
||
|
||
import aiosqlite
|
||
|
||
|
||
async def user_has_role(
|
||
conn: aiosqlite.Connection, *, user_id: int, host_group_id: int, role_name: str
|
||
) -> bool:
|
||
cursor = await conn.execute(
|
||
"""
|
||
SELECT 1 FROM user_hostgroup_roles uhr
|
||
JOIN roles r ON r.id = uhr.role_id
|
||
WHERE uhr.user_id = ?
|
||
AND uhr.host_group_id = ?
|
||
AND r.name = ?
|
||
AND (uhr.expires_at IS NULL OR uhr.expires_at > strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||
LIMIT 1
|
||
""",
|
||
(user_id, host_group_id, role_name),
|
||
)
|
||
row = await cursor.fetchone()
|
||
return row is not None
|
||
|
||
|
||
async def user_has_role_for_host(
|
||
conn: aiosqlite.Connection, *, user_id: int, host_id: int, role_name: str
|
||
) -> bool:
|
||
cursor = await conn.execute("SELECT host_group_id FROM hosts WHERE id = ?", (host_id,))
|
||
row = await cursor.fetchone()
|
||
if row is None:
|
||
return False
|
||
return await user_has_role(conn, user_id=user_id, host_group_id=row[0], role_name=role_name)
|