stephan
63ee0b40b5
- Implemented RuleManager and JSON storage - Added background enforcement scheduler - Added Web UI for rule management at /ui/rules - Bumped version to 0.3.0
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
import asyncio
|
|
import logging
|
|
|
|
from backend import actions
|
|
from backend.auth import is_account_locked, list_manageable_users
|
|
from backend.rules import RuleManager
|
|
from backend.settings import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def check_and_enforce_rules():
|
|
"""Iterate all manageable users and enforce time window rules."""
|
|
settings = get_settings()
|
|
manager = RuleManager(settings)
|
|
|
|
# We only care about users explicitly managed via settings or rules.
|
|
target_users = list_manageable_users(settings)
|
|
|
|
for user in target_users:
|
|
try:
|
|
allowed = manager.is_login_allowed(user)
|
|
rule = manager.get_rule(user)
|
|
|
|
if allowed:
|
|
# If user has a rule with auto-reenable, ensure unlocked
|
|
# We check rule existence because "allowed" is also true for users with NO rules.
|
|
# But for users with no rules, we don't want to auto-unlock randomly (maybe manual lock?).
|
|
# US says: "Given a rule allows automatic reactivation ... Then account is automatically activated"
|
|
# So we only auto-unlock if a rule EXISTS and explicitly asks for it.
|
|
if rule and rule.auto_reenable and is_account_locked(user):
|
|
logger.info("Auto-enabling user %s (Time window started)", user)
|
|
actions.enable_user(user)
|
|
continue
|
|
|
|
# If we are here, access is DENIED.
|
|
|
|
# Check if logged in
|
|
logged_in_users = actions.list_logged_in_users()
|
|
is_logged_in = user in logged_in_users
|
|
|
|
if is_logged_in:
|
|
logger.warning("User %s is logged in during forbidden time. Enforcing logout.", user)
|
|
# Warn and shutdown
|
|
actions.disable_user(
|
|
user,
|
|
countdown=60,
|
|
sound=True,
|
|
message="Time limit reached. Shutdown in 60s.",
|
|
)
|
|
else:
|
|
# Not logged in. Ensure account is locked to prevent login.
|
|
if not is_account_locked(user):
|
|
logger.info("Locking user %s (Time window ended)", user)
|
|
# disable_user locks the account. We pass countdown=0 but since not logged in, it won't matter much.
|
|
actions.disable_user(user, countdown=0, sound=False)
|
|
|
|
except Exception:
|
|
logger.exception("Error enforcing rules for user %s", user)
|
|
|
|
|
|
async def enforcement_loop():
|
|
logger.info("Starting enforcement loop")
|
|
while True:
|
|
try:
|
|
await check_and_enforce_rules()
|
|
except Exception:
|
|
logger.exception("Error in enforcement loop")
|
|
|
|
# Run every minute
|
|
await asyncio.sleep(60)
|