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

added pam auth

Browse Source
This commit is contained in:
Alsan
2026-09-18 17:20:59 +05:00
parent 09bd167536
commit b5593fa7f4
14 changed files with 1154 additions and 1155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+24 -5
View File
@@ -1,12 +1,31 @@
.venv/
# Python
__pycache__/
*.py[cod]
.pytest_cache/
*$py.class
data/*.db
data/*.db-*
# Virtual environments
.venv/
venv/
env/
# Environment / secrets
.env
.env.*
!.env.example
.idea/
# Runtime state
data/state.json
# Logs
*.log
logs/
# IDE / editors
.vscode/
.idea/
*.swp
*.swo
# OS files
.DS_Store
Thumbs.db
README.md
+156 -63
View File
@@ -1,63 +1,156 @@
# Linux Parental Control
A Linux parental-control system for managing Linux user access, daily time allowances, access windows, temporary grants, and automatic session enforcement.
> **Status:** Early development
## Features
- Per-user daily time allowances
- Different allowances for each day of the week
- Multiple access windows per day
- Temporary time grants
- Usage tracking
- Automatic session termination
- Automatic account locking
- Automatic account unlocking
- Reboot-safe enforcement
- Web-based administration
- SQLite database
- systemd service support
---
# Requirements
The application currently targets Linux systems using `systemd`.
You need:
- Linux
- Python 3
- `python-venv`
- `pip`
- `systemd`
- `sudo`
- `passwd`
- `loginctl`
- Git
The enforcement service requires **root privileges** because it manages other Linux users and their sessions.
---
# 1. Clone the Repository
Clone the repository:
```bash
git clone https://git.shihaam.dev/Alsan/linux-user-timer.git
```
## 2. Make it executable
```bash
chmod +x install.sh
```
## 3. Run the install script
```bash
sudo ./install.sh
```
### You can access it via http://127.0.0.1:8765/admin
# Linux Parental Control
A Linux parental-control system for managing Linux user access, daily time allowances, access windows, temporary grants, and automatic session enforcement.
> **Status:** Early development
## Storage
SQLite has been removed.
The application now uses two files under `data/`:
- `config.yaml` — users, daily allowances, access windows, and authentication configuration.
- `state.json` — daily usage, temporary grants, and a bounded event history.
The application never creates or opens `data/parental-control.db`.
`config.yaml` and `state.json` are written atomically and are intended to be root-readable only.
## Web authentication
The administration web interface is protected by **PAM**.
Sign in at:
```text
http://127.0.0.1:8765/admin
```
Use an existing Linux username and its Linux password. The password is passed to PAM for authentication and is not stored by this application.
The PAM service defaults to:
```yaml
auth:
pam_service: login
```
If your distribution uses a different PAM service, change `pam_service` in `data/config.yaml`.
### Restricting who can use the web panel
By default, any Linux account that successfully authenticates through PAM can access the web panel.
For a restricted administration panel, edit `data/config.yaml`:
```yaml
auth:
pam_service: login
admin_users:
- youradminuser
```
Do **not** add a user controlled by the parental-control enforcement system to `admin_users`, because account locking is performed with `passwd`.
The web session is signed with a randomly generated secret stored in:
```text
/etc/parental-control/session-secret
/etc/parental-control/session-secret.env
```
## Features
- Per-user daily time allowances
- Different allowances for each day of the week
- Multiple access windows per day
- Temporary time grants
- Usage tracking
- Automatic session termination
- Automatic account locking
- Automatic account unlocking
- Reboot-safe enforcement
- PAM-authenticated web administration
- YAML configuration
- JSON runtime state
- systemd service support
## Requirements
The application targets Linux systems using `systemd`.
You need:
- Linux
- Python 3
- `python-venv`
- `pip`
- `systemd`
- PAM
- `sudo`
- `passwd`
- `loginctl`
- Git
The enforcement service requires **root privileges** because it manages other Linux users and their sessions.
## Installation
Clone the repository:
```bash
git clone https://git.shihaam.dev/Alsan/linux-user-timer.git
cd linux-user-timer
```
Make the installer executable:
```bash
chmod +x install.sh
```
Run:
```bash
sudo ./install.sh
```
The installer:
1. Installs Python dependencies including PyYAML and python-pam.
2. Creates the Python virtual environment.
3. Removes the legacy `data/parental-control.db` if it exists.
4. Initializes `config.yaml` and `state.json`.
5. Generates a random web-session secret.
6. Installs and starts the systemd service.
## Service commands
```bash
sudo systemctl status parental-control
sudo systemctl restart parental-control
sudo journalctl -u parental-control -f
```
## Reverse proxy / HTTPS
The application listens on:
```text
127.0.0.1:8765
```
Put it behind your existing Nginx/Apache/reverse proxy if you want remote access.
When HTTPS is provided directly to users, set:
```ini
Environment="PARENTAL_CONTROL_HTTPS_ONLY=1"
```
in the systemd service. This marks the session cookie as HTTPS-only.
## Important security note
This application controls Linux accounts and runs as root. Do not expose port `8765` directly to an untrusted network. Prefer binding it to localhost and placing it behind an HTTPS reverse proxy with appropriate firewall rules.
app/database.py
-105
View File
@@ -1,105 +0,0 @@
import sqlite3
from pathlib import Path
from contextlib import contextmanager
BASE_DIR = Path(__file__).resolve().parent.parent
DATABASE_DIR = BASE_DIR / "data"
DATABASE_PATH = DATABASE_DIR / "parental-control.db"
def initialize_database():
DATABASE_DIR.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(DATABASE_PATH) as db:
db.execute("PRAGMA foreign_keys = ON")
db.executescript(
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
enabled INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS daily_allowances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
weekday INTEGER NOT NULL,
allowance_seconds INTEGER NOT NULL DEFAULT 0,
UNIQUE(user_id, weekday),
FOREIGN KEY(user_id)
REFERENCES users(id)
ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS access_windows (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
weekday INTEGER NOT NULL,
start_minute INTEGER NOT NULL,
end_minute INTEGER NOT NULL,
FOREIGN KEY(user_id)
REFERENCES users(id)
ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
date TEXT NOT NULL,
used_seconds INTEGER NOT NULL DEFAULT 0,
UNIQUE(user_id, date),
FOREIGN KEY(user_id)
REFERENCES users(id)
ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS temporary_grants (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
seconds INTEGER NOT NULL,
remaining_seconds INTEGER NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
expires_at TEXT,
consumed INTEGER NOT NULL DEFAULT 0,
FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
event_type TEXT NOT NULL,
details TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(user_id)
REFERENCES users(id)
ON DELETE SET NULL
);
"""
)
db.commit()
@contextmanager
def get_db():
db = sqlite3.connect(DATABASE_PATH)
db.row_factory = sqlite3.Row
db.execute("PRAGMA foreign_keys = ON")
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()
app/enforcement.py
+33 -361
View File
@@ -1,7 +1,14 @@
from datetime import datetime
import subprocess
from .database import get_db
from .storage import (
get_user_policy,
get_remaining_grant_seconds,
consume_grant_seconds,
record_usage,
record_event,
list_users,
)
from .users import (
lock_user,
unlock_user,
@@ -12,11 +19,7 @@ from .users import (
def user_has_session(username: str) -> bool:
result = subprocess.run(
[
"loginctl",
"list-users",
"--no-legend",
],
["loginctl", "list-users", "--no-legend"],
capture_output=True,
text=True,
check=False,
@@ -25,297 +28,28 @@ def user_has_session(username: str) -> bool:
if result.returncode != 0:
return False
for line in result.stdout.splitlines():
parts = line.split()
if len(parts) >= 2 and parts[1] == username:
return True
return False
return any(
len(parts := line.split()) >= 2 and parts[1] == username
for line in result.stdout.splitlines()
)
def current_time():
now = datetime.now()
weekday = now.weekday()
minute = now.hour * 60 + now.minute
return now, weekday, minute
def get_user_policy(user_id: int, weekday: int):
with get_db() as db:
allowance_row = db.execute(
"""
SELECT allowance_seconds
FROM daily_allowances
WHERE user_id = ?
AND weekday = ?
""",
(
user_id,
weekday,
),
).fetchone()
allowance_seconds = (
allowance_row["allowance_seconds"]
if allowance_row
else 0
)
today = datetime.now().date().isoformat()
usage_row = db.execute(
"""
SELECT used_seconds
FROM usage
WHERE user_id = ?
AND date = ?
""",
(
user_id,
today,
),
).fetchone()
usage_seconds = (
usage_row["used_seconds"]
if usage_row
else 0
)
windows = db.execute(
"""
SELECT id, start_minute, end_minute
FROM access_windows
WHERE user_id = ?
AND weekday = ?
ORDER BY start_minute
""",
(
user_id,
weekday,
),
).fetchall()
grant_row = db.execute(
"""
SELECT COALESCE(
SUM(remaining_seconds),
0
) AS total
FROM temporary_grants
WHERE user_id = ?
AND consumed = 0
AND (
expires_at IS NULL
OR expires_at > ?
)
""",
(
user_id,
datetime.now().isoformat(),
),
).fetchone()
grant_seconds = grant_row["total"]
return (
allowance_seconds,
usage_seconds,
windows,
grant_seconds,
)
return now, now.weekday(), now.hour * 60 + now.minute
def is_inside_window(windows, minute: int) -> bool:
if not windows:
return True
for window in windows:
if (
window["start_minute"]
<= minute
< window["end_minute"]
):
return True
return False
return any(
int(window["start_minute"]) <= minute < int(window["end_minute"])
for window in windows
)
def get_remaining_grant_seconds(user_id: int) -> int:
now = datetime.now().isoformat()
with get_db() as db:
row = db.execute(
"""
SELECT COALESCE(
SUM(remaining_seconds),
0
) AS total
FROM temporary_grants
WHERE user_id = ?
AND consumed = 0
AND (
expires_at IS NULL
OR expires_at > ?
)
""",
(
user_id,
now,
),
).fetchone()
return row["total"]
def consume_grant_seconds(
user_id: int,
seconds: int,
):
if seconds <= 0:
return
now = datetime.now().isoformat()
with get_db() as db:
grants = db.execute(
"""
SELECT id, remaining_seconds
FROM temporary_grants
WHERE user_id = ?
AND consumed = 0
AND remaining_seconds > 0
AND (
expires_at IS NULL
OR expires_at > ?
)
ORDER BY id ASC
""",
(
user_id,
now,
),
).fetchall()
remaining = seconds
for grant in grants:
if remaining <= 0:
break
available = grant["remaining_seconds"]
consumed = min(
available,
remaining,
)
new_remaining = (
available - consumed
)
db.execute(
"""
UPDATE temporary_grants
SET remaining_seconds = ?,
consumed = ?
WHERE id = ?
""",
(
new_remaining,
1 if new_remaining <= 0 else 0,
grant["id"],
),
)
remaining -= consumed
def record_usage(
user_id: int,
seconds: int,
):
if seconds <= 0:
return
today = datetime.now().date().isoformat()
with get_db() as db:
row = db.execute(
"""
SELECT used_seconds
FROM usage
WHERE user_id = ?
AND date = ?
""",
(
user_id,
today,
),
).fetchone()
if row is None:
db.execute(
"""
INSERT INTO usage (
user_id,
date,
used_seconds
)
VALUES (?, ?, ?)
""",
(
user_id,
today,
seconds,
),
)
else:
db.execute(
"""
UPDATE usage
SET used_seconds =
used_seconds + ?
WHERE user_id = ?
AND date = ?
""",
(
seconds,
user_id,
today,
),
)
def record_event(
user_id: int,
event_type: str,
details: str = "",
):
with get_db() as db:
db.execute(
"""
INSERT INTO events (
user_id,
event_type,
details
)
VALUES (?, ?, ?)
""",
(
user_id,
event_type,
details,
),
)
def evaluate_user(
user_id: int,
username: str,
):
def evaluate_user(user_id: int, username: str):
now, weekday, minute = current_time()
(
@@ -323,43 +57,16 @@ def evaluate_user(
usage_seconds,
windows,
grant_seconds,
) = get_user_policy(
user_id,
weekday,
)
) = get_user_policy(user_id, weekday)
inside_window = is_inside_window(
windows,
minute,
)
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)
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
)
allowed_by_grant = (
grant_seconds > 0
)
should_allow = (
allowed_by_schedule
or allowed_by_grant
)
allowed_by_schedule = inside_window and allowance_remaining > 0
allowed_by_grant = grant_seconds > 0
should_allow = allowed_by_schedule or allowed_by_grant
locked = is_locked(username)
@@ -367,23 +74,12 @@ def evaluate_user(
if locked:
try:
unlock_user(username)
record_event(
user_id,
"auto_unlock",
"Access became available",
)
record_event(user_id, "auto_unlock", "Access became available")
except Exception as exc:
record_event(
user_id,
"unlock_error",
str(exc),
)
record_event(user_id, "unlock_error", str(exc))
else:
if logged_in:
terminate_user(username)
record_event(
user_id,
"session_terminated",
@@ -393,18 +89,13 @@ def evaluate_user(
if not locked:
try:
lock_user(username)
record_event(
user_id,
"auto_lock",
"Access is not currently permitted",
)
except Exception as exc:
record_event(
user_id,
"lock_error",
str(exc),
)
record_event(user_id, "lock_error", str(exc))
return {
"user_id": user_id,
@@ -424,32 +115,13 @@ def evaluate_user(
def enforce_all_users():
with get_db() as db:
users = db.execute(
"""
SELECT id, username, enabled
FROM users
WHERE enabled = 1
ORDER BY id
"""
).fetchall()
users = list_users()
results = []
for user in users:
try:
result = evaluate_user(
user["id"],
user["username"],
)
results.append(result)
results.append(evaluate_user(user["id"], user["username"]))
except Exception as exc:
record_event(
user["id"],
"enforcement_error",
str(exc),
)
record_event(user["id"], "enforcement_error", str(exc))
return results
app/main.py
+335 -614
View File
File diff suppressed because it is too large Load Diff
app/scheduler.py
+2 -6
View File
@@ -2,12 +2,8 @@ import threading
import time
from datetime import datetime
from .enforcement import (
enforce_all_users,
record_usage,
get_user_policy,
consume_grant_seconds,
)
from .enforcement import enforce_all_users
from .storage import record_usage, get_user_policy, consume_grant_seconds
CHECK_INTERVAL = 5
app/storage.py
+445
View File
@@ -0,0 +1,445 @@
import json
import os
import tempfile
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from threading import RLock
import yaml
BASE_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = BASE_DIR / "data"
CONFIG_PATH = DATA_DIR / "config.yaml"
STATE_PATH = DATA_DIR / "state.json"
_lock = RLock()
DEFAULT_CONFIG = {
"version": 1,
"auth": {
"pam_service": "login",
"admin_users": [],
},
"users": [],
}
DEFAULT_STATE = {
"version": 1,
"next_ids": {
"user": 1,
"window": 1,
"grant": 1,
},
"usage": {},
"temporary_grants": [],
"events": [],
}
def _atomic_write(path: Path, content: str, mode: int = 0o600):
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(
prefix=f".{path.name}.",
dir=str(path.parent),
text=True,
)
try:
os.fchmod(fd, mode)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(content)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp_name, path)
finally:
if os.path.exists(tmp_name):
os.unlink(tmp_name)
def _load_yaml():
if not CONFIG_PATH.exists():
return json.loads(json.dumps(DEFAULT_CONFIG))
with CONFIG_PATH.open("r", encoding="utf-8") as handle:
data = yaml.safe_load(handle) or {}
if not isinstance(data, dict):
raise ValueError("config.yaml must contain a YAML object")
data.setdefault("version", 1)
data.setdefault("auth", {})
data["auth"].setdefault("pam_service", "login")
data["auth"].setdefault("admin_users", [])
data.setdefault("users", [])
return data
def _load_json():
if not STATE_PATH.exists():
return json.loads(json.dumps(DEFAULT_STATE))
with STATE_PATH.open("r", encoding="utf-8") as handle:
data = json.load(handle)
if not isinstance(data, dict):
raise ValueError("state.json must contain a JSON object")
data.setdefault("version", 1)
data.setdefault("next_ids", {})
data["next_ids"].setdefault("user", 1)
data["next_ids"].setdefault("window", 1)
data["next_ids"].setdefault("grant", 1)
data.setdefault("usage", {})
data.setdefault("temporary_grants", [])
data.setdefault("events", [])
return data
def _save_yaml(data):
_atomic_write(
CONFIG_PATH,
yaml.safe_dump(
data,
sort_keys=False,
default_flow_style=False,
allow_unicode=True,
),
)
def _save_json(data):
_atomic_write(
STATE_PATH,
json.dumps(data, indent=2, ensure_ascii=False) + "\n",
)
def initialize_storage():
DATA_DIR.mkdir(parents=True, exist_ok=True)
with _lock:
if not CONFIG_PATH.exists():
_save_yaml(DEFAULT_CONFIG)
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)
_save_json(state)
def get_config():
with _lock:
return _load_yaml()
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 is_admin_allowed(username: str) -> bool:
allowed = admin_users()
return not allowed or username in allowed
def _find_user(config, user_id):
for user in config["users"]:
if int(user["id"]) == int(user_id):
return user
return None
def _find_window(config, window_id):
for user in config["users"]:
for window in user.get("windows", []):
if int(window["id"]) == int(window_id):
return user, window
return None, None
def list_users():
with _lock:
config = _load_yaml()
return sorted(
[dict(user) for user in config["users"] if user.get("enabled", True)],
key=lambda item: item["username"],
)
def get_user(user_id: int):
with _lock:
config = _load_yaml()
user = _find_user(config, user_id)
return dict(user) if user else None
def get_user_by_username(username: str):
with _lock:
config = _load_yaml()
for user in config["users"]:
if user["username"] == username:
return dict(user)
return None
def add_user(username: str):
with _lock:
config = _load_yaml()
if any(u["username"] == username for u in config["users"]):
raise ValueError("User is already configured")
state = _load_json()
user_id = int(state["next_ids"]["user"])
state["next_ids"]["user"] = user_id + 1
user = {
"id": user_id,
"username": username,
"enabled": True,
"allowances": {str(day): 0 for day in range(7)},
"windows": [],
}
config["users"].append(user)
_save_yaml(config)
_save_json(state)
return dict(user)
def delete_user(user_id: int):
with _lock:
config = _load_yaml()
user = _find_user(config, user_id)
if user is None:
return False
username = user["username"]
config["users"] = [
item for item in config["users"]
if int(item["id"]) != int(user_id)
]
state = _load_json()
state["usage"] = {
key: value
for key, value in state["usage"].items()
if not key.startswith(f"{int(user_id)}:")
}
state["temporary_grants"] = [
grant for grant in state["temporary_grants"]
if int(grant["user_id"]) != int(user_id)
]
state["events"] = [
event for event in state["events"]
if event.get("user_id") is None
or int(event["user_id"]) != int(user_id)
]
_save_yaml(config)
_save_json(state)
return username
def set_allowance(user_id: int, weekday: int, seconds: int):
with _lock:
config = _load_yaml()
user = _find_user(config, user_id)
if user is None:
raise KeyError("User not found")
user.setdefault("allowances", {})
user["allowances"][str(weekday)] = int(seconds)
_save_yaml(config)
def add_window(user_id: int, weekday: int, start_minute: int, end_minute: int):
with _lock:
config = _load_yaml()
user = _find_user(config, user_id)
if user is None:
raise KeyError("User not found")
state = _load_json()
window_id = int(state["next_ids"]["window"])
state["next_ids"]["window"] = window_id + 1
user.setdefault("windows", []).append({
"id": window_id,
"weekday": int(weekday),
"start_minute": int(start_minute),
"end_minute": int(end_minute),
})
_save_yaml(config)
_save_json(state)
return window_id
def delete_window(window_id: int):
with _lock:
config = _load_yaml()
owner, window = _find_window(config, window_id)
if owner is None:
return None
owner["windows"] = [
item for item in owner.get("windows", [])
if int(item["id"]) != int(window_id)
]
_save_yaml(config)
return int(owner["id"])
def get_user_policy(user_id: int, weekday: int):
with _lock:
config = _load_yaml()
user = _find_user(config, user_id)
if user is None:
return 0, 0, [], 0
allowance_seconds = int(
user.get("allowances", {}).get(str(weekday), 0)
)
windows = sorted(
[
dict(window)
for window in user.get("windows", [])
if int(window["weekday"]) == int(weekday)
],
key=lambda item: int(item["start_minute"]),
)
state = _load_json()
today = datetime.now().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
)
)
return allowance_seconds, usage_seconds, windows, grant_seconds
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
)
)
def add_grant(user_id: int, seconds: int):
with _lock:
state = _load_json()
grant_id = int(state["next_ids"]["grant"])
state["next_ids"]["grant"] = grant_id + 1
state["temporary_grants"].append({
"id": grant_id,
"user_id": int(user_id),
"seconds": int(seconds),
"remaining_seconds": int(seconds),
"created_at": datetime.now().isoformat(timespec="seconds"),
"expires_at": None,
"consumed": False,
})
_save_json(state)
return grant_id
def consume_grant_seconds(user_id: int, seconds: int):
if seconds <= 0:
return
with _lock:
state = _load_json()
now = datetime.now().isoformat()
remaining = int(seconds)
for grant in state["temporary_grants"]:
if remaining <= 0:
break
if int(grant["user_id"]) != int(user_id):
continue
if grant.get("consumed", False):
continue
if int(grant["remaining_seconds"]) <= 0:
continue
if (
grant.get("expires_at") is not None
and grant["expires_at"] <= now
):
continue
available = int(grant["remaining_seconds"])
consumed = min(available, remaining)
new_remaining = available - consumed
grant["remaining_seconds"] = new_remaining
grant["consumed"] = new_remaining <= 0
remaining -= consumed
_save_json(state)
def record_usage(user_id: int, seconds: int):
if seconds <= 0:
return
with _lock:
state = _load_json()
today = datetime.now().date().isoformat()
key = f"{int(user_id)}:{today}"
state["usage"][key] = int(state["usage"].get(key, 0)) + int(seconds)
_save_json(state)
def list_grants(user_id: int, limit: int = 20):
with _lock:
state = _load_json()
grants = [
dict(grant)
for grant in state["temporary_grants"]
if int(grant["user_id"]) == int(user_id)
]
grants.sort(key=lambda item: int(item["id"]), reverse=True)
return grants[:limit]
def record_event(user_id, event_type: str, details: str = ""):
with _lock:
state = _load_json()
state["events"].append({
"user_id": int(user_id) if user_id is not None else None,
"event_type": event_type,
"details": details,
"created_at": datetime.now().isoformat(timespec="seconds"),
})
# Keep the state file bounded.
state["events"] = state["events"][-2000:]
_save_json(state)
data/config.yaml
+6
View File
@@ -0,0 +1,6 @@
version: 1
auth:
pam_service: login
admin_users:
- root
users: []
install.sh
+40
View File
@@ -156,6 +156,7 @@ case "$PACKAGE_MANAGER" in
esac
# ============================================================
# Verify Python
# ============================================================
@@ -200,6 +201,27 @@ echo "[3/5] Installing Python packages"
-r "$INSTALL_DIR/requirements.txt"
# ============================================================
# Remove legacy SQLite storage
# ============================================================
LEGACY_DB="$INSTALL_DIR/data/parental-control.db"
if [ -f "$LEGACY_DB" ]; then
echo
echo "Removing legacy SQLite database:"
echo " $LEGACY_DB"
rm -f "$LEGACY_DB"
fi
mkdir -p "$INSTALL_DIR/data"
chmod 700 "$INSTALL_DIR/data"
# Keep the configuration and runtime state readable/writable only by root.
[ -f "$INSTALL_DIR/data/config.yaml" ] && chmod 600 "$INSTALL_DIR/data/config.yaml"
[ -f "$INSTALL_DIR/data/state.json" ] && chmod 600 "$INSTALL_DIR/data/state.json"
# ============================================================
# Install systemd service
# ============================================================
@@ -208,6 +230,23 @@ echo
echo "[4/5] Installing systemd service"
SERVICE_FILE="/etc/systemd/system/parental-control.service"
SECRET_DIR="/etc/parental-control"
SECRET_FILE="$SECRET_DIR/session-secret"
SECRET_ENV_FILE="$SECRET_DIR/session-secret.env"
install -d -m 700 "$SECRET_DIR"
if [ ! -s "$SECRET_FILE" ]; then
"$INSTALL_DIR/.venv/bin/python" -c \
'import secrets; print(secrets.token_urlsafe(48))' \
> "$SECRET_FILE"
chmod 600 "$SECRET_FILE"
fi
# systemd reads the environment file; the raw secret is kept separately
# so it is easy to rotate without changing the service definition.
printf 'PARENTAL_CONTROL_SESSION_SECRET=%s\n' "$(cat "$SECRET_FILE")" > "$SECRET_ENV_FILE"
chmod 600 "$SECRET_ENV_FILE"
sed \
"s|%INSTALL_DIR%|$INSTALL_DIR|g" \
@@ -215,6 +254,7 @@ sed \
> "$SERVICE_FILE"
# ============================================================
# Enable and start service
# ============================================================
parental-control.service
+2 -1
View File
@@ -9,8 +9,9 @@ WorkingDirectory=%INSTALL_DIR%
Environment="PATH=%INSTALL_DIR%/.venv/bin:/usr/local/sbin:/usr/local/bin:/usr/bin:/usr/sbin:/sbin:/bin"
Environment="PARENTAL_CONTROL_ENFORCEMENT=1"
EnvironmentFile=-/etc/parental-control/session-secret.env
ExecStart=%INSTALL_DIR%/.venv/bin/uvicorn app.main:app --host 127.0.0.1 --port 8765
ExecStart=%INSTALL_DIR%/.venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8765
Restart=always
RestartSec=5
requirements.txt
+3
View File
@@ -7,10 +7,13 @@ h11==0.16.0
idna==3.19
Jinja2==3.1.6
MarkupSafe==3.0.3
python-pam>=2.0.2
pydantic==2.13.5
pydantic_core==2.46.5
PyYAML>=6.0
python-multipart==0.0.32
starlette==1.6.0
typing-inspection==0.4.4
typing_extensions==4.16.0
uvicorn==0.53.0
itsdangerous
templates/index.html
+10
View File
@@ -254,6 +254,14 @@
Parental Control
</h1>
<div style="margin-top:8px;font-size:14px;">Signed in as <strong>{{ username }}</strong>
<form method="post" action="/logout" style="display:inline;margin-left:12px;">
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button type="submit" class="button-secondary">Log out</button>
</form>
</div>
</header>
@@ -345,6 +353,7 @@
method="post"
action="/admin/users/{{ user.id }}/delete"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button
class="button-danger"
@@ -401,6 +410,7 @@
method="post"
action="/admin/users"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<input
templates/login.html
+82
View File
@@ -0,0 +1,82 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Sign in — Parental Control</title>
<style>
* { box-sizing: border-box; }
body {
margin: 0;
min-height: 100vh;
display: grid;
place-items: center;
background: #f4f5f7;
color: #1f2937;
font-family: system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
}
.card {
width: min(420px, calc(100% - 32px));
background: white;
border-radius: 12px;
padding: 28px;
box-shadow: 0 2px 12px rgba(0,0,0,.10);
}
h1 { margin: 0 0 8px; }
p { color: #6b7280; }
label { display:block; margin-top:16px; font-weight:600; font-size:14px; }
input {
width:100%;
margin-top:7px;
border:1px solid #d1d5db;
border-radius:7px;
padding:10px 12px;
font-size:15px;
}
button {
width:100%;
margin-top:22px;
border:0;
border-radius:7px;
padding:11px 15px;
background:#2563eb;
color:white;
cursor:pointer;
font-size:15px;
}
.error {
background:#fef2f2;
color:#991b1b;
border:1px solid #fecaca;
border-radius:7px;
padding:10px 12px;
margin-top:16px;
}
.hint { font-size:13px; }
</style>
</head>
<body>
<main class="card">
<h1>Parental Control</h1>
<p>Sign in with a Linux account authenticated through PAM.</p>
{% if error %}
<div class="error">{{ error }}</div>
{% endif %}
<form method="post" action="/login">
<input type="hidden" name="next" value="{{ next }}">
<label for="username">Linux username</label>
<input id="username" name="username" type="text" autocomplete="username" required autofocus>
<label for="password">Password</label>
<input id="password" name="password" type="password" autocomplete="current-password" required>
<button type="submit">Sign in</button>
</form>
<p class="hint">The application does not store your Linux password.</p>
</main>
</body>
</html>
templates/user.html
+16
View File
@@ -302,6 +302,14 @@
Parental Control
</h1>
<div style="margin-top:8px;font-size:14px;">Signed in as <strong>{{ username }}</strong>
<form method="post" action="/logout" style="display:inline;margin-left:12px;">
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button type="submit" class="secondary">Log out</button>
</form>
</div>
</header>
<main>
@@ -359,6 +367,7 @@
method="post"
action="/admin/users/{{ user.id }}/unlock"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button
class="primary"
@@ -375,6 +384,7 @@
method="post"
action="/admin/users/{{ user.id }}/lock"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button
class="warning"
@@ -392,6 +402,7 @@
method="post"
action="/admin/users/{{ user.id }}/terminate"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button
class="secondary"
@@ -476,6 +487,7 @@
method="post"
action="/admin/users/{{ user.id }}/allowance"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<input
type="hidden"
@@ -613,6 +625,7 @@
method="post"
action="/admin/windows/{{ window.id }}/delete"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button
class="danger"
@@ -634,6 +647,7 @@
action="/admin/users/{{ user.id }}/window"
style="margin-top: 12px;"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<input
type="hidden"
@@ -686,6 +700,7 @@
method="post"
action="/admin/users/{{ user.id }}/grant"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<input
type="number"
@@ -841,6 +856,7 @@
);
"
>
<input type="hidden" name="csrf" value="{{ csrf_token }}">
<button
class="danger"