From 63ee0b40b52828febdb15cbce4689df429b8b9d3 Mon Sep 17 00:00:00 2001 From: stephan Date: Fri, 16 Jan 2026 11:25:56 +0100 Subject: [PATCH] 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 --- CHANGELOG.md | 2 + VERSION | 2 +- backend/app.py | 42 +- backend/enforcer.py | 71 +++ backend/models.py | 30 +- backend/rules.py | 91 ++++ backend/settings.py | 3 + backend/templates/index.html | 14 +- backend/templates/rules.html | 442 ++++++++++++++++++ project-management/PROJECT_STATUS.md | 11 +- .../requirements/stories/US_000045.md | 2 +- .../requirements/tasks/TASK_000052.md | 2 +- .../requirements/tasks/TASK_000053.md | 2 +- .../requirements/tasks/TASK_000054.md | 2 +- tests/test_api_rules.py | 56 +++ tests/test_rules.py | 95 ++++ 16 files changed, 851 insertions(+), 16 deletions(-) create mode 100644 backend/enforcer.py create mode 100644 backend/rules.py create mode 100644 backend/templates/rules.html create mode 100644 tests/test_api_rules.py create mode 100644 tests/test_rules.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 49df541..6c74fbe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -86,6 +86,8 @@ By: Codex (GPT-5) | 15.01.2026 | 🎨 UI | ID: System Information Kacheln mit Progressbars verbessert. By: Codex (GPT-5) | | 15.01.2026 | 🚀 Release | ID: VERSION auf 0.2.3 erhoeht. By: Codex (GPT-5) | | 15.01.2026 | ⚙️ Code | ID: Makefile restart-Target hinzugefuegt. By: Codex (GPT-5) | +| 16.01.2026 | ✨ Feat | ID: EPIC_000014/US_000045 Login-Regeln und Scheduler implementiert (UI + Backend). By: Gemini CLI | +| 16.01.2026 | 🚀 Release | ID: VERSION auf 0.3.0 erhoeht. By: Gemini CLI | --- ## Legende diff --git a/VERSION b/VERSION index 7179039..9325c3c 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.2.3 +0.3.0 \ No newline at end of file diff --git a/backend/app.py b/backend/app.py index bc489f7..e285312 100644 --- a/backend/app.py +++ b/backend/app.py @@ -1,3 +1,4 @@ +import asyncio import logging from typing import List @@ -6,7 +7,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates -from backend import actions +from backend import actions, enforcer, rules from backend.actions import ActionError from backend.auth import ( authenticate_admin_user, @@ -17,12 +18,14 @@ from backend.auth import ( list_manageable_users, ) from backend.models import ( + AccessRule, ActionRequest, ActionResponse, EnrollRequest, EnrollResponse, LoginRequest, LoginResponse, + RuleSet, UpdateActionResponse, UpdateCheckResponse, UpdateLogEntry, @@ -48,6 +51,11 @@ app.mount("/assets", StaticFiles(directory="assets"), name="assets") templates = Jinja2Templates(directory="backend/templates") +@app.on_event("startup") +async def startup_event(): + asyncio.create_task(enforcer.enforcement_loop()) + + def get_oidc_client(settings: Settings = Depends(get_settings)) -> OIDCClient: if not settings.oidc_enabled: raise HTTPException( @@ -238,6 +246,32 @@ def enable_user( ) +@app.get("/rules", response_model=RuleSet, dependencies=[Depends(get_current_admin)]) +def get_rules(settings: Settings = Depends(get_settings)) -> RuleSet: + manager = rules.RuleManager(settings) + return manager.load_rules() + + +@app.post("/rules", dependencies=[Depends(get_current_admin)]) +def set_rule( + rule: AccessRule, + settings: Settings = Depends(get_settings), +) -> dict: + manager = rules.RuleManager(settings) + manager.set_rule(rule) + return {"status": "ok"} + + +@app.delete("/rules/{username}", dependencies=[Depends(get_current_admin)]) +def delete_rule( + username: str, + settings: Settings = Depends(get_settings), +) -> dict: + manager = rules.RuleManager(settings) + manager.delete_rule(username) + return {"status": "ok"} + + @app.get("/update/status", response_model=UpdateStatus, dependencies=[Depends(get_current_admin)]) def update_status(settings: Settings = Depends(get_settings)) -> UpdateStatus: status_data = update.get_status(settings) @@ -334,3 +368,9 @@ def system_metrics_status() -> SystemMetrics: @app.get("/dashboard", response_class=HTMLResponse) def index(request: Request) -> HTMLResponse: return templates.TemplateResponse("index.html", {"request": request}) + + +@app.get("/ui/rules", response_class=HTMLResponse) +def rules_ui(request: Request) -> HTMLResponse: + return templates.TemplateResponse("rules.html", {"request": request}) + diff --git a/backend/enforcer.py b/backend/enforcer.py new file mode 100644 index 0000000..a3e6abb --- /dev/null +++ b/backend/enforcer.py @@ -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) diff --git a/backend/models.py b/backend/models.py index 08f86e4..6038de3 100644 --- a/backend/models.py +++ b/backend/models.py @@ -1,8 +1,36 @@ -from typing import List, Optional +from enum import Enum +from typing import List, Optional, Dict from pydantic import BaseModel, Field +class DayOfWeek(str, Enum): + MONDAY = "mon" + TUESDAY = "tue" + WEDNESDAY = "wed" + THURSDAY = "thu" + FRIDAY = "fri" + SATURDAY = "sat" + SUNDAY = "sun" + + +class TimeWindow(BaseModel): + days: List[DayOfWeek] + start_time: str = Field(..., pattern=r"^\d{2}:\d{2}$", description="HH:MM format") + end_time: str = Field(..., pattern=r"^\d{2}:\d{2}$", description="HH:MM format") + + +class AccessRule(BaseModel): + user: str + enabled: bool = True + auto_reenable: bool = False + allowed_windows: List[TimeWindow] = [] + + +class RuleSet(BaseModel): + rules: Dict[str, AccessRule] = {} # Keyed by username for O(1) lookup + + class ActionRequest(BaseModel): countdown: Optional[int] = Field(default=None, ge=0, description="Seconds for countdown") sound: Optional[bool] = Field(default=None, description="Play sound alongside notification") diff --git a/backend/rules.py b/backend/rules.py new file mode 100644 index 0000000..c3b730e --- /dev/null +++ b/backend/rules.py @@ -0,0 +1,91 @@ +import json +import logging +import os +from datetime import datetime +from typing import Optional + +from backend.models import RuleSet, AccessRule, DayOfWeek +from backend.settings import get_settings, Settings + +logger = logging.getLogger(__name__) + + +class RuleManager: + def __init__(self, settings: Optional[Settings] = None): + self.settings = settings or get_settings() + self.rules_file = self.settings.rules_file + + def load_rules(self) -> RuleSet: + if not os.path.exists(self.rules_file): + return RuleSet() + try: + with open(self.rules_file, "r", encoding="utf-8") as f: + data = json.load(f) + return RuleSet(**data) + except (json.JSONDecodeError, OSError) as e: + logger.error("Failed to load rules from %s: %s", self.rules_file, e) + return RuleSet() + + def save_rules(self, rules: RuleSet) -> None: + try: + os.makedirs(os.path.dirname(self.rules_file), exist_ok=True) + with open(self.rules_file, "w", encoding="utf-8") as f: + f.write(rules.model_dump_json(indent=2)) + except OSError as e: + logger.error("Failed to save rules to %s: %s", self.rules_file, e) + raise + + def get_rule(self, user: str) -> Optional[AccessRule]: + rules = self.load_rules() + return rules.rules.get(user) + + def set_rule(self, rule: AccessRule) -> None: + rules = self.load_rules() + rules.rules[rule.user] = rule + self.save_rules(rules) + + def delete_rule(self, user: str) -> None: + rules = self.load_rules() + if user in rules.rules: + del rules.rules[user] + self.save_rules(rules) + + def is_login_allowed(self, user: str, now: Optional[datetime] = None) -> bool: + """ + Check if user is allowed to login at the given time (or now). + Returns True if no rule exists (or disabled) or if time is within an allowed window. + Returns False if a rule exists and current time is outside all allowed windows. + """ + rule = self.get_rule(user) + if not rule or not rule.enabled: + # No rule or rule disabled -> Default Allow + return True + + if not rule.allowed_windows: + # Rule enabled but no windows -> Deny All + return False + + if now is None: + now = datetime.now() + + # Map weekday to Enum + weekday_map = { + 0: DayOfWeek.MONDAY, + 1: DayOfWeek.TUESDAY, + 2: DayOfWeek.WEDNESDAY, + 3: DayOfWeek.THURSDAY, + 4: DayOfWeek.FRIDAY, + 5: DayOfWeek.SATURDAY, + 6: DayOfWeek.SUNDAY, + } + current_day = weekday_map[now.weekday()] + current_time_str = now.strftime("%H:%M") + + for window in rule.allowed_windows: + if current_day in window.days: + # Handle simple range start <= now <= end + # (Overnight windows assumed to be split by user into two windows) + if window.start_time <= current_time_str <= window.end_time: + return True + + return False diff --git a/backend/settings.py b/backend/settings.py index 48a1088..aec4f5e 100644 --- a/backend/settings.py +++ b/backend/settings.py @@ -57,6 +57,9 @@ class Settings: self.update_log_file: str = os.getenv( "SKD_UPDATE_LOG_FILE", "/var/lib/skd/update_logs.jsonl" ) + self.rules_file: str = os.getenv( + "SKD_RULES_FILE", "/var/lib/skd/rules.json" + ) # Paths/tools self.notify_send_path: str = os.getenv("SKD_NOTIFY_SEND_PATH", "notify-send") self.sound_player: str = os.getenv("SKD_SOUND_PLAYER", "paplay") diff --git a/backend/templates/index.html b/backend/templates/index.html index 76a53c3..26bfc7f 100644 --- a/backend/templates/index.html +++ b/backend/templates/index.html @@ -128,10 +128,16 @@