feat: login time window rules (EPIC_000014)
- Implemented RuleManager and JSON storage - Added background enforcement scheduler - Added Web UI for rule management at /ui/rules - Bumped version to 0.3.0
This commit is contained in:
71
backend/enforcer.py
Normal file
71
backend/enforcer.py
Normal file
@ -0,0 +1,71 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user