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
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
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
|