102 lines
3 KiB
Python
102 lines
3 KiB
Python
|
"""
|
||
|
|
File-store operations for local-identity.
|
||
|
|
|
||
|
|
The store lives at ~/.local-identity/ by default. Set LOCAL_IDENTITY_HOME
|
||
|
|
to override (useful for tests and for running multiple independent stores).
|
||
|
|
|
||
|
|
Directory layout:
|
||
|
|
$LOCAL_IDENTITY_HOME/
|
||
|
|
├── config.yaml operator email and optional overrides (mode 600)
|
||
|
|
└── users/
|
||
|
|
├── <user>.yaml primary user (mode 600)
|
||
|
|
├── <user>1.yaml test user 1 (mode 600)
|
||
|
|
└── <user>2.yaml test user 2 (mode 600)
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import List
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
from .user import UserRecord
|
||
|
|
|
||
|
|
|
||
|
|
def _store_dir() -> Path:
|
||
|
|
"""Return the store root directory. Overridable via LOCAL_IDENTITY_HOME."""
|
||
|
|
return Path(os.environ.get("LOCAL_IDENTITY_HOME", str(Path.home() / ".local-identity")))
|
||
|
|
|
||
|
|
|
||
|
|
def _users_dir() -> Path:
|
||
|
|
return _store_dir() / "users"
|
||
|
|
|
||
|
|
|
||
|
|
def _config_file() -> Path:
|
||
|
|
return _store_dir() / "config.yaml"
|
||
|
|
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------ #
|
||
|
|
# Directory management #
|
||
|
|
# ------------------------------------------------------------------ #
|
||
|
|
|
||
|
|
def store_exists() -> bool:
|
||
|
|
return _store_dir().exists()
|
||
|
|
|
||
|
|
|
||
|
|
def init_dirs() -> None:
|
||
|
|
"""Create the store directory tree with secure permissions."""
|
||
|
|
store = _store_dir()
|
||
|
|
users = _users_dir()
|
||
|
|
store.mkdir(mode=0o700, exist_ok=True)
|
||
|
|
users.mkdir(mode=0o700, exist_ok=True)
|
||
|
|
# Enforce permissions even when dirs already existed
|
||
|
|
os.chmod(store, 0o700)
|
||
|
|
os.chmod(users, 0o700)
|
||
|
|
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------ #
|
||
|
|
# User CRUD #
|
||
|
|
# ------------------------------------------------------------------ #
|
||
|
|
|
||
|
|
def write_user(user: UserRecord) -> None:
|
||
|
|
path = _users_dir() / f"{user.username}.yaml"
|
||
|
|
path.write_text(user.to_yaml(), encoding="utf-8")
|
||
|
|
os.chmod(path, 0o600)
|
||
|
|
|
||
|
|
|
||
|
|
def read_user(username: str) -> UserRecord:
|
||
|
|
path = _users_dir() / f"{username}.yaml"
|
||
|
|
if not path.exists():
|
||
|
|
raise FileNotFoundError(f"User '{username}' not found in store")
|
||
|
|
return UserRecord.from_yaml(path.read_text(encoding="utf-8"))
|
||
|
|
|
||
|
|
|
||
|
|
def list_users() -> List[UserRecord]:
|
||
|
|
users_dir = _users_dir()
|
||
|
|
if not users_dir.exists():
|
||
|
|
return []
|
||
|
|
return [
|
||
|
|
UserRecord.from_yaml(p.read_text(encoding="utf-8"))
|
||
|
|
for p in sorted(users_dir.glob("*.yaml"))
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------ #
|
||
|
|
# Config #
|
||
|
|
# ------------------------------------------------------------------ #
|
||
|
|
|
||
|
|
def read_config() -> dict:
|
||
|
|
config_file = _config_file()
|
||
|
|
if not config_file.exists():
|
||
|
|
return {}
|
||
|
|
return yaml.safe_load(config_file.read_text(encoding="utf-8")) or {}
|
||
|
|
|
||
|
|
|
||
|
|
def write_config(config: dict) -> None:
|
||
|
|
config_file = _config_file()
|
||
|
|
config_file.write_text(
|
||
|
|
yaml.dump(config, default_flow_style=False, allow_unicode=True),
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
os.chmod(config_file, 0o600)
|