Alsan/linux-user-timer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added graps

Browse Source
This commit is contained in:
Alsan
2026-09-19 12:46:23 +05:00
parent b5593fa7f4
commit bd973a641c
16 changed files with 735 additions and 1508 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/enforcement.py
+9 -7
View File
@@ -3,9 +3,6 @@ import subprocess
from .storage import (
get_user_policy,
get_remaining_grant_seconds,
consume_grant_seconds,
record_usage,
record_event,
list_users,
)
@@ -40,6 +37,7 @@ def current_time():
def is_inside_window(windows, minute: int) -> bool:
# No configured windows means that no time-of-day restriction exists.
if not windows:
return True
@@ -57,14 +55,17 @@ def evaluate_user(user_id: int, username: str):
usage_seconds,
windows,
grant_seconds,
) = get_user_policy(user_id, weekday)
) = get_user_policy(user_id, weekday, now.date())
inside_window = is_inside_window(windows, minute)
allowance_remaining = max(0, allowance_seconds - usage_seconds)
total_remaining = allowance_remaining + grant_seconds
logged_in = user_has_session(username)
allowed_by_schedule = inside_window and allowance_remaining > 0
# A user's normal allowance is always constrained by the configured
# access window (when one exists). A temporary grant bypasses the window
# and normal allowance by design.
normal_access_available = allowance_remaining > 0
allowed_by_schedule = inside_window and normal_access_available
allowed_by_grant = grant_seconds > 0
should_allow = allowed_by_schedule or allowed_by_grant
@@ -104,12 +105,13 @@ def evaluate_user(user_id: int, username: str):
"weekday": weekday,
"minute": minute,
"inside_window": inside_window,
"has_configured_windows": bool(windows),
"logged_in": logged_in,
"allowance_seconds": allowance_seconds,
"usage_seconds": usage_seconds,
"allowance_remaining": allowance_remaining,
"grant_seconds": grant_seconds,
"total_remaining": total_remaining,
"normal_access_available": normal_access_available,
"allowed": should_allow,
}
app/main.py
+52 -10
View File
@@ -1,3 +1,4 @@
import os
import secrets
from pathlib import Path
@@ -23,9 +24,12 @@ from .storage import (
add_grant,
list_grants,
is_admin_allowed,
get_usage_history,
)
from .users import (
linux_user_exists,
is_non_root_user,
list_available_users,
lock_user,
unlock_user,
terminate_user,
@@ -34,13 +38,13 @@ from .users import (
BASE_DIR = Path(__file__).resolve().parent.parent
SESSION_SECRET = __import__("os").environ.get(
SESSION_SECRET = os.environ.get(
"PARENTAL_CONTROL_SESSION_SECRET"
) or secrets.token_urlsafe(32)
app = FastAPI(
title="Parental Control",
version="0.2.0",
version="0.3.0",
)
app.add_middleware(
@@ -49,9 +53,7 @@ app.add_middleware(
session_cookie="parental_control_session",
max_age=8 * 60 * 60,
same_site="lax",
https_only=__import__("os").environ.get(
"PARENTAL_CONTROL_HTTPS_ONLY", "0"
) == "1",
https_only=os.environ.get("PARENTAL_CONTROL_HTTPS_ONLY", "0") == "1",
)
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
@@ -83,6 +85,8 @@ def current_user(request: Request):
if not username:
return None
# Re-check the Linux PAM group on every request so removing an account
# from the admin group takes effect without waiting for the session TTL.
if not is_admin_allowed(username):
request.session.clear()
return None
@@ -100,7 +104,6 @@ def require_web_auth(request: Request):
f"/login?next={next_path}",
status_code=303,
)
return None
@@ -144,16 +147,19 @@ class GrantRequest(BaseModel):
def root():
return {
"application": "Parental Control",
"version": "0.2.0",
"version": "0.3.0",
"status": "running",
"authentication": "PAM",
"authentication": "PAM + pam Linux group",
}
@app.get("/login")
def login_page(request: Request, next: str = "/admin"):
if current_user(request):
return RedirectResponse(next if next.startswith("/") and not next.startswith("//") else "/admin", status_code=303)
return RedirectResponse(
next if next.startswith("/") and not next.startswith("//") else "/admin",
status_code=303,
)
return templates.TemplateResponse(
request=request,
@@ -161,6 +167,7 @@ def login_page(request: Request, next: str = "/admin"):
context={
"next": next if next.startswith("/") else "/admin",
"error": None,
"pam_group": get_config()["auth"].get("pam_group", "pam"),
},
)
@@ -191,6 +198,7 @@ def login(
context={
"next": safe_next,
"error": "Invalid Linux username or password.",
"pam_group": get_config()["auth"].get("pam_group", "pam"),
},
status_code=401,
)
@@ -201,7 +209,8 @@ def login(
name="login.html",
context={
"next": safe_next,
"error": "This Linux account is not allowed to access the administration panel.",
"error": "Your Linux account is authenticated, but it is not a member of the PAM administration group.",
"pam_group": get_config()["auth"].get("pam_group", "pam"),
},
status_code=403,
)
@@ -236,6 +245,14 @@ def list_users_api(request: Request):
]
@app.get("/api/users/{user_id}/usage")
def user_usage_api(request: Request, user_id: int, days: int = 14):
require_api_auth(request)
if get_user(user_id) is None:
raise HTTPException(status_code=404, detail="User not found")
return get_usage_history(user_id, days)
@app.post("/api/users/{user_id}/lock")
def manually_lock(user_id: int, request: Request):
require_api_auth(request)
@@ -318,13 +335,21 @@ def admin_page(request: Request):
if redirect:
return redirect
configured = {user["username"] for user in list_users()}
available_users = [
user for user in list_available_users()
if user["username"] not in configured
]
return templates.TemplateResponse(
request=request,
name="index.html",
context={
"users": list_users(),
"available_users": available_users,
"username": current_user(request),
"csrf_token": csrf_token(request),
"pam_group": get_config()["auth"].get("pam_group", "pam"),
},
)
@@ -349,6 +374,12 @@ def admin_user_page(request: Request, user_id: int):
key=lambda item: (int(item["weekday"]), int(item["start_minute"])),
)
usage_history = get_usage_history(user_id, 14)
total_used = sum(item["used_seconds"] for item in usage_history)
total_allowance = sum(item["allowance_seconds"] for item in usage_history)
today_used = usage_history[-1]["used_seconds"] if usage_history else 0
today_allowance = usage_history[-1]["allowance_seconds"] if usage_history else 0
return templates.TemplateResponse(
request=request,
name="user.html",
@@ -359,6 +390,11 @@ def admin_user_page(request: Request, user_id: int):
"allowances": allowances,
"windows": windows,
"grants": list_grants(user_id),
"usage_history": usage_history,
"total_used": total_used,
"total_allowance": total_allowance,
"today_used": today_used,
"today_allowance": today_allowance,
"csrf_token": csrf_token(request),
"username": current_user(request),
},
@@ -531,6 +567,12 @@ def admin_add_user(
if not linux_user_exists(username):
raise HTTPException(status_code=400, detail="Linux user does not exist")
if not is_non_root_user(username):
raise HTTPException(
status_code=400,
detail="Only regular non-root Linux users can be added to parental control.",
)
if get_user_by_username(username) is not None:
raise HTTPException(status_code=400, detail="User is already configured")
app/scheduler.py
+35 -108
View File
@@ -1,156 +1,86 @@
import threading
import time
from datetime import datetime
from datetime import date
from .enforcement import enforce_all_users
from .storage import record_usage, get_user_policy, consume_grant_seconds
from .storage import get_user_policy, record_usage, consume_grant_seconds
CHECK_INTERVAL = 5
class Scheduler:
def __init__(
self,
interval: int = CHECK_INTERVAL,
):
def __init__(self, interval: int = CHECK_INTERVAL):
self.interval = interval
self._thread = None
self._stop_event = threading.Event()
self._last_usage_update = {}
def start(self):
if (
self._thread is not None
and self._thread.is_alive()
):
if self._thread is not None and self._thread.is_alive():
return
self._stop_event.clear()
self._thread = threading.Thread(
target=self._run,
name="parental-control-scheduler",
daemon=True,
)
self._thread.start()
def stop(self):
self._stop_event.set()
if self._thread is not None:
self._thread.join(
timeout=self.interval + 2
)
self._thread.join(timeout=self.interval + 2)
def _run(self):
# Evaluate immediately when the
# application starts.
self._tick()
while not self._stop_event.wait(
self.interval
):
while not self._stop_event.wait(self.interval):
self._tick()
def _tick(self):
now = time.monotonic()
results = enforce_all_users()
for result in results:
user_id = result["user_id"]
previous = self._last_usage_update.get(user_id)
if not result["logged_in"]:
self._last_usage_update.pop(
user_id,
None,
)
continue
if previous is not None:
elapsed = max(0, int(now - previous["monotonic"]))
if elapsed > 0 and previous["allowed"] and previous["logged_in"]:
self._record_allowed_usage(
user_id,
elapsed,
previous["weekday"],
previous["date"],
)
if not result["allowed"]:
self._last_usage_update.pop(
user_id,
None,
)
continue
previous = (
self._last_usage_update.get(
user_id
)
)
self._last_usage_update[user_id] = now
if previous is None:
continue
elapsed = int(
now - previous
)
if elapsed <= 0:
continue
self._record_allowed_usage(
user_id,
elapsed,
)
def _record_allowed_usage(
self,
user_id: int,
seconds: int,
):
if result["logged_in"] and result["allowed"]:
self._last_usage_update[user_id] = {
"monotonic": now,
"allowed": True,
"logged_in": True,
"weekday": result["weekday"],
"date": result["timestamp"][:10],
}
else:
self._last_usage_update.pop(user_id, None)
def _record_allowed_usage(self, user_id: int, seconds: int, weekday: int, usage_date: str):
if seconds <= 0:
return
weekday = datetime.now().weekday()
(
allowance_seconds,
usage_seconds,
windows,
_windows,
grant_seconds,
) = get_user_policy(
user_id,
weekday,
)
) = get_user_policy(user_id, weekday)
allowance_remaining = max(
0,
allowance_seconds
- usage_seconds,
)
normal_usage = min(
seconds,
allowance_remaining,
)
grant_usage = (
seconds
- normal_usage
)
if grant_usage > grant_seconds:
grant_usage = grant_seconds
total_usage = (
normal_usage
+ grant_usage
)
allowance_remaining = max(0, allowance_seconds - usage_seconds)
normal_usage = min(seconds, allowance_remaining)
grant_usage = min(max(0, seconds - normal_usage), grant_seconds)
total_usage = normal_usage + grant_usage
if total_usage <= 0:
return
@@ -158,14 +88,11 @@ class Scheduler:
record_usage(
user_id,
total_usage,
date.fromisoformat(usage_date),
)
if grant_usage > 0:
consume_grant_seconds(
user_id,
grant_usage,
)
consume_grant_seconds(user_id, grant_usage)
scheduler = Scheduler()
app/storage.py
+75 -40
View File
@@ -1,10 +1,9 @@
import json
import os
import tempfile
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from datetime import datetime, date, timedelta
from threading import RLock
from pathlib import Path
import yaml
@@ -19,7 +18,7 @@ DEFAULT_CONFIG = {
"version": 1,
"auth": {
"pam_service": "login",
"admin_users": [],
"pam_group": "pam",
},
"users": [],
}
@@ -68,9 +67,13 @@ def _load_yaml():
data.setdefault("version", 1)
data.setdefault("auth", {})
if not isinstance(data["auth"], dict):
raise ValueError("config.yaml auth must be an object")
data["auth"].setdefault("pam_service", "login")
data["auth"].setdefault("admin_users", [])
data["auth"].setdefault("pam_group", "pam")
data.setdefault("users", [])
if not isinstance(data["users"], list):
raise ValueError("config.yaml users must be a list")
return data
@@ -124,8 +127,6 @@ def initialize_storage():
if not STATE_PATH.exists():
_save_json(DEFAULT_STATE)
# Keep files usable after manual edits while avoiding destructive
# initialization or recreation of any database.
config = _load_yaml()
state = _load_json()
_save_yaml(config)
@@ -141,14 +142,14 @@ def get_pam_service():
return get_config()["auth"].get("pam_service", "login")
def admin_users():
value = get_config()["auth"].get("admin_users", [])
return {str(item) for item in value}
def get_pam_group():
return get_config()["auth"].get("pam_group", "pam")
def is_admin_allowed(username: str) -> bool:
allowed = admin_users()
return not allowed or username in allowed
"""Keep authorization in users.py so PAM group membership is system-backed."""
from .users import user_in_group
return user_in_group(username, get_pam_group())
def _find_user(config, user_id):
@@ -256,7 +257,7 @@ def set_allowance(user_id: int, weekday: int, seconds: int):
raise KeyError("User not found")
user.setdefault("allowances", {})
user["allowances"][str(weekday)] = int(seconds)
user["allowances"][str(weekday)] = max(0, int(seconds))
_save_yaml(config)
@@ -298,7 +299,26 @@ def delete_window(window_id: int):
return int(owner["id"])
def get_user_policy(user_id: int, weekday: int):
def _active_grant_seconds(state, user_id: int, now_iso: str) -> int:
return sum(
int(grant["remaining_seconds"])
for grant in state["temporary_grants"]
if int(grant["user_id"]) == int(user_id)
and not grant.get("consumed", False)
and int(grant.get("remaining_seconds", 0)) > 0
and (
grant.get("expires_at") is None
or grant["expires_at"] > now_iso
)
)
def get_user_policy(user_id: int, weekday: int, on_date: date | None = None):
"""Return the complete policy for a specific weekday/date.
Configuration is always read from the current files, so allowance and
access-window changes are order-independent.
"""
with _lock:
config = _load_yaml()
user = _find_user(config, user_id)
@@ -318,22 +338,14 @@ def get_user_policy(user_id: int, weekday: int):
)
state = _load_json()
today = datetime.now().date().isoformat()
target_date = on_date or datetime.now().date()
today = target_date.isoformat()
usage_seconds = int(
state["usage"].get(f"{int(user_id)}:{today}", 0)
)
now = datetime.now().isoformat()
grant_seconds = sum(
int(grant["remaining_seconds"])
for grant in state["temporary_grants"]
if int(grant["user_id"]) == int(user_id)
and not grant.get("consumed", False)
and (
grant.get("expires_at") is None
or grant["expires_at"] > now
)
)
grant_seconds = _active_grant_seconds(state, user_id, now)
return allowance_seconds, usage_seconds, windows, grant_seconds
@@ -341,17 +353,10 @@ def get_user_policy(user_id: int, weekday: int):
def get_remaining_grant_seconds(user_id: int) -> int:
with _lock:
state = _load_json()
now = datetime.now().isoformat()
return sum(
int(grant["remaining_seconds"])
for grant in state["temporary_grants"]
if int(grant["user_id"]) == int(user_id)
and not grant.get("consumed", False)
and int(grant["remaining_seconds"]) > 0
and (
grant.get("expires_at") is None
or grant["expires_at"] > now
)
return _active_grant_seconds(
state,
user_id,
datetime.now().isoformat(),
)
@@ -407,18 +412,49 @@ def consume_grant_seconds(user_id: int, seconds: int):
_save_json(state)
def record_usage(user_id: int, seconds: int):
def record_usage(user_id: int, seconds: int, usage_date: date | None = None):
if seconds <= 0:
return
with _lock:
state = _load_json()
today = datetime.now().date().isoformat()
key = f"{int(user_id)}:{today}"
target_date = usage_date or datetime.now().date()
key = f"{int(user_id)}:{target_date.isoformat()}"
state["usage"][key] = int(state["usage"].get(key, 0)) + int(seconds)
_save_json(state)
def get_usage_history(user_id: int, days: int = 14):
days = max(1, min(int(days), 90))
with _lock:
config = _load_yaml()
state = _load_json()
user = _find_user(config, user_id)
if user is None:
return []
today = datetime.now().date()
history = []
for offset in range(days - 1, -1, -1):
day = today - timedelta(days=offset)
weekday = day.weekday()
allowance = int(
user.get("allowances", {}).get(str(weekday), 0)
)
key = f"{int(user_id)}:{day.isoformat()}"
used = int(state["usage"].get(key, 0))
history.append({
"date": day.isoformat(),
"weekday": weekday,
"used_seconds": used,
"allowance_seconds": allowance,
})
return history
def list_grants(user_id: int, limit: int = 20):
with _lock:
state = _load_json()
@@ -440,6 +476,5 @@ def record_event(user_id, event_type: str, details: str = ""):
"details": details,
"created_at": datetime.now().isoformat(timespec="seconds"),
})
# Keep the state file bounded.
state["events"] = state["events"][-2000:]
_save_json(state)
app/users.py
+114
View File
@@ -1,5 +1,7 @@
import grp
import pwd
import subprocess
from typing import List, Dict
def linux_user_exists(username: str) -> bool:
@@ -14,6 +16,118 @@ def get_uid(username: str) -> int:
return pwd.getpwnam(username).pw_uid
def _login_def_value(name: str, default: int) -> int:
"""Read an integer value from /etc/login.defs."""
try:
with open("/etc/login.defs", "r", encoding="utf-8") as handle:
for line in handle:
line = line.strip()
if not line or line.startswith("#"):
continue
parts = line.split()
if len(parts) >= 2 and parts[0] == name:
value = int(parts[1])
if value > 0:
return value
except (OSError, ValueError):
pass
return default
def _uid_min() -> int:
return _login_def_value("UID_MIN", 1000)
def _uid_max() -> int:
return _login_def_value("UID_MAX", 60000)
def _is_interactive_shell(shell: str) -> bool:
"""Exclude service accounts that cannot be used for interactive logins."""
shell = (shell or "").strip().lower()
if not shell:
return False
return not shell.endswith(("/nologin", "/false"))
def _supplementary_groups(username: str) -> set[str]:
groups = set()
try:
user = pwd.getpwnam(username)
except KeyError:
return groups
try:
groups.add(grp.getgrgid(user.pw_gid).gr_name)
except KeyError:
pass
for group in grp.getgrall():
if username in group.gr_mem:
groups.add(group.gr_name)
return groups
def has_root_privileges(username: str) -> bool:
"""Best-effort detection for accounts with ordinary root-style group access."""
try:
user = pwd.getpwnam(username)
except KeyError:
return False
if user.pw_uid == 0:
return True
groups = _supplementary_groups(username)
return bool(groups.intersection({"root", "wheel", "sudo"}))
def is_non_root_user(username: str) -> bool:
"""Return True for regular non-root accounts suitable for parental control."""
try:
user = pwd.getpwnam(username)
except KeyError:
return False
if user.pw_uid < _uid_min() or user.pw_uid > _uid_max():
return False
if username in {"nobody", "nfsnobody"}:
return False
if not _is_interactive_shell(user.pw_shell):
return False
return not has_root_privileges(username)
def list_available_users() -> List[Dict[str, str]]:
"""Return regular interactive users that can be selected for parental control."""
users = []
for user in pwd.getpwall():
if not is_non_root_user(user.pw_name):
continue
users.append({"username": user.pw_name})
return sorted(users, key=lambda item: item["username"].lower())
def user_in_group(username: str, group_name: str) -> bool:
try:
group = grp.getgrnam(group_name)
user = pwd.getpwnam(username)
except KeyError:
return False
return (
username in group.gr_mem
or user.pw_gid == group.gr_gid
)
def is_locked(username: str) -> bool:
result = subprocess.run(
["passwd", "-S", username],