azcomputerguru/claudetools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: auto-sync from HOWARD-HOME at 2026-07-05 19:59:52

Browse Source 
Author: Howard Enos
Machine: HOWARD-HOME
Timestamp: 2026-07-05 19:59:52
This commit is contained in:
Howard Enos
2026-07-05 20:00:20 -07:00
parent b0ebd44c9d
commit 976bfe7957
12 changed files with 1712 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

436
.claude/skills/owncloud/scripts/owncloud.py
View File

@@ -17,6 +17,10 @@ from __future__ import annotations
import argparse
import json
import os
import re
import secrets
import string
import sys
from pathlib import Path
@@ -26,6 +30,9 @@ import backup_common as bc # noqa: E402
GB = 1_000_000_000
# occ enforces this uid charset (a-z A-Z 0-9 and + _ . @ - ').
UID_RE = re.compile(r"^[A-Za-z0-9+_.@'\-]+$")
def _gb(n) -> float:
try:
@@ -34,6 +41,43 @@ def _gb(n) -> float:
return 0.0
# --- write-op gating (mirrors the fleet convention: refuse without --confirm) --
def _gated(desc: str, confirm: bool) -> bool:
if not confirm:
print("[WARNING] Refusing state-changing action without --confirm.")
print(f"[INFO] Would: {desc}")
return False
return True
def _gen_password(n: int = 20) -> str:
alpha = string.ascii_letters + string.digits + "!@#%^*-_=+"
return "".join(secrets.choice(alpha) for _ in range(n))
def _resolve_password(args, *, allow_gen: bool = True):
"""Return (password, was_generated) from --password / --password-env / --gen-password.
Prefer --password-env or --gen-password: a literal --password lands in shell
history and `ps` on the operator's machine."""
if getattr(args, "password", None):
return args.password, False
env_var = getattr(args, "password_env", None)
if env_var:
v = os.environ.get(env_var)
if not v:
raise SystemExit(f"[ERROR] env var {env_var} is unset/empty")
return v, False
if allow_gen and getattr(args, "gen_password", False):
return _gen_password(), True
return None, False
def _vault_reminder(uid: str) -> None:
print("[CRITICAL] House rule: store this credential in the SOPS vault now, e.g.:")
print(f" bash .claude/scripts/vault.sh set-field "
f"infrastructure/owncloud-users/{uid}.sops.yaml credentials.password '<paste-here>'")
def cmd_status(c: OwncloudClient, args) -> int:
s = c.server_status()
print(f"[OK] {c.host}")
@@ -132,6 +176,321 @@ def cmd_audit(c: OwncloudClient, args) -> int:
return 0
# ---------------------------------------------------------------------------
# Management commands (writes gated behind --confirm; reads are ungated)
# ---------------------------------------------------------------------------
def _ok(msg: str, payload: dict, as_json: bool) -> int:
if as_json:
print(json.dumps(payload, indent=2))
else:
print(f"[OK] {msg}")
return 0
# -- user lifecycle ---------------------------------------------------------
def cmd_user_add(c: OwncloudClient, args) -> int:
if not UID_RE.match(args.uid):
print(f"[ERROR] invalid uid {args.uid!r}: allowed chars are "
"a-z A-Z 0-9 + _ . @ - '", file=sys.stderr)
return 2
pw, generated = _resolve_password(args)
if not pw:
print("[ERROR] supply --password, --password-env VAR, or --gen-password",
file=sys.stderr)
return 2
desc = f"create ownCloud user '{args.uid}'"
if args.group:
desc += f" in group(s) {', '.join(args.group)}"
if not _gated(desc, args.confirm):
return 3
res = c.create_user(args.uid, pw, args.display_name, args.email, args.group)
print(f"[OK] created user '{args.uid}'")
if generated:
print(f"[CRITICAL] generated password (shown once): {pw}")
_vault_reminder(args.uid)
if args.json:
print(json.dumps({"uid": args.uid, "created": True,
"generated_password": generated, "detail": res}, indent=2))
return 0
def cmd_user_delete(c: OwncloudClient, args) -> int:
if not _gated(f"DELETE user '{args.uid}' and ALL their data"
+ (" (forced)" if args.force else ""), args.confirm):
return 3
out = c.delete_user(args.uid, force=args.force)
return _ok(f"deleted user '{args.uid}'", {"uid": args.uid, "deleted": True,
"detail": out}, args.json)
def cmd_user_enable(c: OwncloudClient, args) -> int:
if not _gated(f"enable user '{args.uid}'", args.confirm):
return 3
out = c.set_user_enabled(args.uid, True)
return _ok(f"enabled '{args.uid}'", {"uid": args.uid, "enabled": True, "detail": out}, args.json)
def cmd_user_disable(c: OwncloudClient, args) -> int:
if not _gated(f"disable user '{args.uid}'", args.confirm):
return 3
out = c.set_user_enabled(args.uid, False)
return _ok(f"disabled '{args.uid}'", {"uid": args.uid, "enabled": False, "detail": out}, args.json)
def cmd_user_reset_password(c: OwncloudClient, args) -> int:
pw, generated = _resolve_password(args)
if not pw:
print("[ERROR] supply --password, --password-env VAR, or --gen-password",
file=sys.stderr)
return 2
if not _gated(f"reset password for user '{args.uid}'", args.confirm):
return 3
c.reset_password(args.uid, pw)
print(f"[OK] reset password for '{args.uid}'")
if generated:
print(f"[CRITICAL] generated password (shown once): {pw}")
_vault_reminder(args.uid)
if args.json:
print(json.dumps({"uid": args.uid, "password_reset": True,
"generated_password": generated}, indent=2))
return 0
def cmd_user_modify(c: OwncloudClient, args) -> int:
changes = []
if args.display_name is not None:
changes.append(("displayname", args.display_name))
if args.email is not None:
changes.append(("email", args.email))
if not changes:
print("[ERROR] nothing to change: pass --display-name and/or --email",
file=sys.stderr)
return 2
desc = "set " + ", ".join(f"{k}={v!r}" for k, v in changes) + f" on '{args.uid}'"
if not _gated(desc, args.confirm):
return 3
for key, val in changes:
c.modify_user(args.uid, key, val)
return _ok(f"modified '{args.uid}'", {"uid": args.uid,
"changed": dict(changes)}, args.json)
# -- quota ------------------------------------------------------------------
def cmd_quota(c: OwncloudClient, args) -> int:
"""Read a user's quota (ungated)."""
q = c.get_quota(args.uid)
return _ok(f"{args.uid} quota: {q or 'default'}",
{"uid": args.uid, "quota": q}, args.json)
def cmd_quota_set(c: OwncloudClient, args) -> int:
if not _gated(f"set quota for '{args.uid}' to {args.value!r}", args.confirm):
return 3
c.set_quota(args.uid, args.value)
return _ok(f"set '{args.uid}' quota = {args.value}",
{"uid": args.uid, "quota": args.value}, args.json)
def cmd_quota_reset(c: OwncloudClient, args) -> int:
if not _gated(f"reset '{args.uid}' quota to default", args.confirm):
return 3
c.reset_quota(args.uid)
return _ok(f"reset '{args.uid}' quota to default",
{"uid": args.uid, "quota": "default"}, args.json)
# -- groups -----------------------------------------------------------------
def cmd_groups(c: OwncloudClient, args) -> int:
g = c.list_groups()
if args.json:
print(json.dumps(g, indent=2))
return 0
names = g if isinstance(g, list) else list(g.keys())
for n in names:
print(f" {n}")
print(f"\n{len(names)} groups")
return 0
def cmd_group_members(c: OwncloudClient, args) -> int:
m = c.group_members(args.group)
if args.json:
print(json.dumps(m, indent=2))
return 0
members = m if isinstance(m, list) else list(m.keys())
for u in members:
print(f" {u}")
print(f"\n{len(members)} members of '{args.group}'")
return 0
def cmd_user_groups(c: OwncloudClient, args) -> int:
g = c.user_groups(args.uid)
if args.json:
print(json.dumps(g, indent=2))
return 0
groups = g if isinstance(g, list) else list(g.keys())
for n in groups:
print(f" {n}")
print(f"\n'{args.uid}' is in {len(groups)} group(s)")
return 0
def cmd_group_add(c: OwncloudClient, args) -> int:
if not _gated(f"create group '{args.group}'", args.confirm):
return 3
c.create_group(args.group)
return _ok(f"created group '{args.group}'", {"group": args.group, "created": True}, args.json)
def cmd_group_delete(c: OwncloudClient, args) -> int:
if not _gated(f"delete group '{args.group}'", args.confirm):
return 3
c.delete_group(args.group)
return _ok(f"deleted group '{args.group}'", {"group": args.group, "deleted": True}, args.json)
def cmd_group_add_member(c: OwncloudClient, args) -> int:
if not _gated(f"add {', '.join(args.member)} to group '{args.group}'", args.confirm):
return 3
c.add_group_members(args.group, args.member)
return _ok(f"added {len(args.member)} member(s) to '{args.group}'",
{"group": args.group, "added": args.member}, args.json)
def cmd_group_remove_member(c: OwncloudClient, args) -> int:
if not _gated(f"remove {', '.join(args.member)} from group '{args.group}'", args.confirm):
return 3
c.remove_group_members(args.group, args.member)
return _ok(f"removed {len(args.member)} member(s) from '{args.group}'",
{"group": args.group, "removed": args.member}, args.json)
# -- shares (read-only) -----------------------------------------------------
def cmd_shares(c: OwncloudClient, args) -> int:
shares = c.list_shares(args.user)
if args.json:
print(json.dumps(shares, indent=2))
return 0
print(f"{'ID':>7} {'TYPE':12} {'OWNER':16} {'SHARE WITH':16} {'PATH/TARGET':30}")
for s in shares:
print(f"{str(s['id']):>7} {s['share_type'][:12]:12} "
f"{str(s['owner'] or '-')[:16]:16} {str(s['share_with'] or '-')[:16]:16} "
f"{str(s['path'] or s['target'] or '-')[:30]:30}")
print(f"\n{len(shares)} share(s)"
+ (f" involving '{args.user}'" if args.user else ""))
return 0
# -- server / apps / config -------------------------------------------------
def cmd_maintenance(c: OwncloudClient, args) -> int:
if args.state == "status":
on = c.maintenance_status()
return _ok(f"maintenance mode is {'ON' if on else 'off'}",
{"maintenance": on}, args.json)
want_on = args.state == "on"
if not _gated(f"turn maintenance mode {'ON' if want_on else 'off'}", args.confirm):
return 3
c.maintenance_mode(want_on)
return _ok(f"maintenance mode {'ON' if want_on else 'off'}",
{"maintenance": want_on}, args.json)
def cmd_apps(c: OwncloudClient, args) -> int:
apps = c.list_apps()
if args.json:
print(json.dumps(apps, indent=2))
return 0
enabled = apps.get("enabled", {}) if isinstance(apps, dict) else {}
disabled = apps.get("disabled", {}) if isinstance(apps, dict) else {}
print(f"--- enabled ({len(enabled)}) ---")
for a in sorted(enabled):
print(f" {a}")
print(f"--- disabled ({len(disabled)}) ---")
for a in sorted(disabled):
print(f" {a}")
return 0
def cmd_app_enable(c: OwncloudClient, args) -> int:
if not _gated(f"enable app '{args.app}'", args.confirm):
return 3
out = c.enable_app(args.app)
return _ok(f"enabled app '{args.app}'", {"app": args.app, "enabled": True, "detail": out}, args.json)
def cmd_app_disable(c: OwncloudClient, args) -> int:
if not _gated(f"disable app '{args.app}'", args.confirm):
return 3
out = c.disable_app(args.app)
return _ok(f"disabled app '{args.app}'", {"app": args.app, "enabled": False, "detail": out}, args.json)
def cmd_config_get(c: OwncloudClient, args) -> int:
if args.name:
val = c.config_get(args.name)
return _ok(f"{args.name} = {val}", {"name": args.name, "value": val}, args.json)
cfg = c.config_list()
print(json.dumps(cfg, indent=2))
return 0
def cmd_config_set(c: OwncloudClient, args) -> int:
if not _gated(f"set system config {args.name} = {args.value!r} ({args.type})", args.confirm):
return 3
c.config_set(args.name, args.value, args.type)
return _ok(f"set {args.name} = {args.value}",
{"name": args.name, "value": args.value, "type": args.type}, args.json)
# -- files admin ------------------------------------------------------------
def cmd_files_scan(c: OwncloudClient, args) -> int:
if not args.all and not args.user:
print("[ERROR] specify a user or --all", file=sys.stderr)
return 2
target = "ALL users" if args.all else f"'{args.user}'"
if not _gated(f"filesystem rescan for {target}"
+ (f" path={args.path}" if args.path else ""), args.confirm):
return 3
out = c.files_scan(uid=args.user, all_users=args.all, path=args.path)
return _ok(f"scanned {target}", {"scanned": target, "detail": out.splitlines()[-1:] if out else []}, args.json)
def cmd_transfer_ownership(c: OwncloudClient, args) -> int:
if not _gated(f"transfer files from '{args.source}' to '{args.dest}'"
+ (f" path={args.path}" if args.path else ""), args.confirm):
return 3
out = c.transfer_ownership(args.source, args.dest, args.path)
return _ok(f"transferred '{args.source}' -> '{args.dest}'",
{"source": args.source, "dest": args.dest, "detail": out}, args.json)
def _cleanup(c: OwncloudClient, args, kind: str) -> int:
"""Shared guard for trashbin/versions cleanup: refuse a fleet-wide sweep
unless it is made explicit with --all-users (occ treats 'no user' as ALL)."""
users = args.user or []
if not users and not args.all_users:
print(f"[ERROR] {kind} cleanup with no --user targets ALL users; "
"pass explicit --user, or --all-users to sweep everyone.",
file=sys.stderr)
return 2
target = "ALL users" if args.all_users else ", ".join(users)
if not _gated(f"permanently delete {kind} for {target}", args.confirm):
return 3
fn = c.trashbin_cleanup if kind == "trashbin" else c.versions_cleanup
out = fn([] if args.all_users else users)
return _ok(f"{kind} cleaned for {target}",
{"cleanup": kind, "target": target, "detail": out}, args.json)
def cmd_trashbin_cleanup(c: OwncloudClient, args) -> int:
return _cleanup(c, args, "trashbin")
def cmd_versions_cleanup(c: OwncloudClient, args) -> int:
return _cleanup(c, args, "versions")
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(prog="owncloud", description="ownCloud backup inventory (GPS audit)")
p.add_argument("--host", help="override server host/IP (default 172.16.3.22)")
@@ -141,6 +500,62 @@ def build_parser() -> argparse.ArgumentParser:
u = sub.add_parser("users"); u.add_argument("--json", action="store_true")
us = sub.add_parser("usage"); us.add_argument("--json", action="store_true")
a = sub.add_parser("audit"); a.add_argument("--client"); a.add_argument("--rmm", action="store_true"); a.add_argument("--json", action="store_true")
def _pw_opts(sp):
sp.add_argument("--password", help="password literal (avoid: lands in shell history/ps)")
sp.add_argument("--password-env", metavar="VAR", help="read password from this env var")
sp.add_argument("--gen-password", action="store_true", help="generate a strong password")
# --- user lifecycle (writes) ---
ua = sub.add_parser("user-add")
ua.add_argument("uid"); ua.add_argument("--display-name"); ua.add_argument("--email")
ua.add_argument("-g", "--group", action="append", help="group to add to (repeatable)")
_pw_opts(ua); ua.add_argument("--confirm", action="store_true"); ua.add_argument("--json", action="store_true")
ud = sub.add_parser("user-delete"); ud.add_argument("uid")
ud.add_argument("-f", "--force", action="store_true"); ud.add_argument("--confirm", action="store_true"); ud.add_argument("--json", action="store_true")
ue = sub.add_parser("user-enable"); ue.add_argument("uid"); ue.add_argument("--confirm", action="store_true"); ue.add_argument("--json", action="store_true")
udi = sub.add_parser("user-disable"); udi.add_argument("uid"); udi.add_argument("--confirm", action="store_true"); udi.add_argument("--json", action="store_true")
urp = sub.add_parser("user-reset-password"); urp.add_argument("uid")
_pw_opts(urp); urp.add_argument("--confirm", action="store_true"); urp.add_argument("--json", action="store_true")
um = sub.add_parser("user-modify"); um.add_argument("uid")
um.add_argument("--display-name"); um.add_argument("--email")
um.add_argument("--confirm", action="store_true"); um.add_argument("--json", action="store_true")
# --- quota ---
q = sub.add_parser("quota"); q.add_argument("uid"); q.add_argument("--json", action="store_true")
qs = sub.add_parser("quota-set"); qs.add_argument("uid"); qs.add_argument("value", help="e.g. '5 GB', 'none' (unlimited)")
qs.add_argument("--confirm", action="store_true"); qs.add_argument("--json", action="store_true")
qr = sub.add_parser("quota-reset"); qr.add_argument("uid"); qr.add_argument("--confirm", action="store_true"); qr.add_argument("--json", action="store_true")
# --- groups ---
gl = sub.add_parser("groups"); gl.add_argument("--json", action="store_true")
gm = sub.add_parser("group-members"); gm.add_argument("group"); gm.add_argument("--json", action="store_true")
ug = sub.add_parser("user-groups"); ug.add_argument("uid"); ug.add_argument("--json", action="store_true")
gadd = sub.add_parser("group-add"); gadd.add_argument("group"); gadd.add_argument("--confirm", action="store_true"); gadd.add_argument("--json", action="store_true")
gdel = sub.add_parser("group-delete"); gdel.add_argument("group"); gdel.add_argument("--confirm", action="store_true"); gdel.add_argument("--json", action="store_true")
gam = sub.add_parser("group-add-member"); gam.add_argument("group"); gam.add_argument("-m", "--member", action="append", required=True); gam.add_argument("--confirm", action="store_true"); gam.add_argument("--json", action="store_true")
grm = sub.add_parser("group-remove-member"); grm.add_argument("group"); grm.add_argument("-m", "--member", action="append", required=True); grm.add_argument("--confirm", action="store_true"); grm.add_argument("--json", action="store_true")
# --- shares (read-only) ---
sh = sub.add_parser("shares"); sh.add_argument("--user"); sh.add_argument("--json", action="store_true")
# --- server / apps / config ---
mnt = sub.add_parser("maintenance"); mnt.add_argument("state", choices=["status", "on", "off"]); mnt.add_argument("--confirm", action="store_true"); mnt.add_argument("--json", action="store_true")
ap = sub.add_parser("apps"); ap.add_argument("--json", action="store_true")
ape = sub.add_parser("app-enable"); ape.add_argument("app"); ape.add_argument("--confirm", action="store_true"); ape.add_argument("--json", action="store_true")
apd = sub.add_parser("app-disable"); apd.add_argument("app"); apd.add_argument("--confirm", action="store_true"); apd.add_argument("--json", action="store_true")
cg = sub.add_parser("config-get"); cg.add_argument("name", nargs="?", help="omit to dump all system config"); cg.add_argument("--json", action="store_true")
cs = sub.add_parser("config-set"); cs.add_argument("name"); cs.add_argument("--value", required=True); cs.add_argument("--type", default="string", choices=["string", "integer", "double", "boolean", "json"]); cs.add_argument("--confirm", action="store_true"); cs.add_argument("--json", action="store_true")
# --- files admin ---
fs = sub.add_parser("files-scan"); fs.add_argument("--user"); fs.add_argument("--all", action="store_true"); fs.add_argument("--path"); fs.add_argument("--confirm", action="store_true"); fs.add_argument("--json", action="store_true")
to = sub.add_parser("transfer-ownership"); to.add_argument("source"); to.add_argument("dest"); to.add_argument("--path"); to.add_argument("--confirm", action="store_true"); to.add_argument("--json", action="store_true")
tc = sub.add_parser("trashbin-cleanup"); tc.add_argument("--user", action="append"); tc.add_argument("--all-users", action="store_true"); tc.add_argument("--confirm", action="store_true"); tc.add_argument("--json", action="store_true")
vc = sub.add_parser("versions-cleanup"); vc.add_argument("--user", action="append"); vc.add_argument("--all-users", action="store_true"); vc.add_argument("--confirm", action="store_true"); vc.add_argument("--json", action="store_true")
return p
@@ -148,7 +563,26 @@ def main(argv=None) -> int:
args = build_parser().parse_args(argv)
c = OwncloudClient(args.host)
handlers = {"status": cmd_status, "users": cmd_users,
"usage": cmd_usage, "audit": cmd_audit}
"usage": cmd_usage, "audit": cmd_audit,
"user-add": cmd_user_add, "user-delete": cmd_user_delete,
"user-enable": cmd_user_enable, "user-disable": cmd_user_disable,
"user-reset-password": cmd_user_reset_password,
"user-modify": cmd_user_modify,
"quota": cmd_quota, "quota-set": cmd_quota_set,
"quota-reset": cmd_quota_reset,
"groups": cmd_groups, "group-members": cmd_group_members,
"user-groups": cmd_user_groups,
"group-add": cmd_group_add, "group-delete": cmd_group_delete,
"group-add-member": cmd_group_add_member,
"group-remove-member": cmd_group_remove_member,
"shares": cmd_shares,
"maintenance": cmd_maintenance, "apps": cmd_apps,
"app-enable": cmd_app_enable, "app-disable": cmd_app_disable,
"config-get": cmd_config_get, "config-set": cmd_config_set,
"files-scan": cmd_files_scan,
"transfer-ownership": cmd_transfer_ownership,
"trashbin-cleanup": cmd_trashbin_cleanup,
"versions-cleanup": cmd_versions_cleanup}
try:
return handlers[args.cmd](c, args)
except bc.BackupError as exc:

204
.claude/skills/owncloud/scripts/owncloud_client.py
View File

@@ -227,6 +227,210 @@ class OwncloudClient:
u["used_source"] = "du" if du is not None else None
return users
# -- occ helpers for management ops ----------------------------------------
def occ_env(self, env: dict, args: str,
timeout: int = EXEC_TIMEOUT) -> tuple[int, str, str]:
"""Run occ with extra environment (e.g. OC_PASS) exported for the apache
user. `sudo -u apache env KEY=val php occ ...` — the explicit `env` sets
the var inside apache's context regardless of sudo's env_reset policy.
Caveat: the value is briefly visible in `ps` on the server to root; this
is occ's own documented mechanism for passwords and the box is root-only.
"""
prefix = " ".join(f"{k}={shlex.quote(str(v))}" for k, v in env.items())
cmd = f"sudo -u apache env {prefix} php {OCC_INSTALL_DIR}/occ {args}"
return self.run(cmd, timeout=timeout)
def _occ_ok(self, args: str, timeout: int = EXEC_TIMEOUT) -> str:
"""Run occ, raise BackupError on non-zero, return stdout."""
rc, out, err = self.occ(args, timeout=timeout)
if rc != 0:
label = args.split()[0] if args else "occ"
raise bc.BackupError(f"occ {label} failed (rc={rc}): "
f"{(err or out).strip()[:300]}")
return out
# -- user lifecycle --------------------------------------------------------
def create_user(self, uid: str, password: str,
display_name: Optional[str] = None,
email: Optional[str] = None,
groups: Optional[list[str]] = None) -> dict:
parts = [f"user:add {shlex.quote(uid)} --password-from-env -n"]
if display_name:
parts.append(f"--display-name={shlex.quote(display_name)}")
if email:
parts.append(f"--email={shlex.quote(email)}")
for g in (groups or []):
parts.append(f"-g {shlex.quote(g)}")
rc, out, err = self.occ_env({"OC_PASS": password}, " ".join(parts))
if rc != 0:
raise bc.BackupError(f"user:add {uid} failed (rc={rc}): "
f"{(err or out).strip()[:300]}")
return {"uid": uid, "output": out.strip()}
def delete_user(self, uid: str, force: bool = False) -> str:
return self._occ_ok(f"user:delete {shlex.quote(uid)}"
+ (" -f" if force else ""), timeout=DU_TIMEOUT).strip()
def set_user_enabled(self, uid: str, enabled: bool) -> str:
verb = "user:enable" if enabled else "user:disable"
return self._occ_ok(f"{verb} {shlex.quote(uid)}").strip()
def reset_password(self, uid: str, password: str) -> str:
rc, out, err = self.occ_env(
{"OC_PASS": password},
f"user:resetpassword {shlex.quote(uid)} --password-from-env")
if rc != 0:
raise bc.BackupError(f"user:resetpassword {uid} failed (rc={rc}): "
f"{(err or out).strip()[:300]}")
return out.strip()
def modify_user(self, uid: str, key: str, value: str) -> str:
"""key is one of: displayname, email (per occ user:modify)."""
return self._occ_ok(
f"user:modify {shlex.quote(uid)} {shlex.quote(key)} {shlex.quote(value)}").strip()
# -- quota (occ user:setting <uid> files quota) ----------------------------
def get_quota(self, uid: str) -> Optional[str]:
rc, out, err = self.occ(f"user:setting {shlex.quote(uid)} files quota")
return out.strip() or None if rc == 0 else None
def set_quota(self, uid: str, value: str) -> str:
return self._occ_ok(
f"user:setting {shlex.quote(uid)} files quota --value={shlex.quote(value)}").strip()
def reset_quota(self, uid: str) -> str:
return self._occ_ok(
f"user:setting {shlex.quote(uid)} files quota --delete").strip()
# -- groups ----------------------------------------------------------------
def list_groups(self) -> Any:
return json.loads(self._occ_ok("group:list --output=json") or "[]")
def group_members(self, group: str) -> Any:
return json.loads(self._occ_ok(
f"group:list-members {shlex.quote(group)} --output=json") or "[]")
def user_groups(self, uid: str) -> Any:
return json.loads(self._occ_ok(
f"user:list-groups {shlex.quote(uid)} --output=json") or "[]")
def create_group(self, name: str) -> str:
return self._occ_ok(f"group:add {shlex.quote(name)}").strip()
def delete_group(self, name: str) -> str:
return self._occ_ok(f"group:delete {shlex.quote(name)}").strip()
def add_group_members(self, group: str, members: list[str]) -> str:
ms = " ".join(f"-m {shlex.quote(m)}" for m in members)
return self._occ_ok(f"group:add-member {shlex.quote(group)} {ms}").strip()
def remove_group_members(self, group: str, members: list[str]) -> str:
ms = " ".join(f"-m {shlex.quote(m)}" for m in members)
return self._occ_ok(f"group:remove-member {shlex.quote(group)} {ms}").strip()
# -- shares (read-only, straight from the DB; occ has no share list) --------
SHARE_TYPES = {0: "user", 1: "group", 3: "public_link",
4: "email", 6: "federated"}
def list_shares(self, uid: Optional[str] = None) -> list[dict]:
# NB: uid is NOT inlined into SQL. MySQL honours backslash escapes by
# default, so ''-doubling would not safely contain a hostile/odd uid
# (and occ allows "'" in uids). Fetch all rows, filter in Python.
pfx = self._db_config()["dbtableprefix"]
sql = (f"SELECT s.id, s.share_type, s.share_with, s.uid_owner, "
f"s.file_target, s.permissions, s.stime, fc.path "
f"FROM {pfx}share s "
f"LEFT JOIN {pfx}filecache fc ON fc.fileid=s.file_source "
f"ORDER BY s.stime DESC;")
out = self._mysql(sql)
shares: list[dict] = []
for row in out.splitlines():
f = [None if v in ("NULL", "") else v for v in row.split("\t")]
if len(f) < 8:
continue
if uid and uid not in (f[2], f[3]): # share_with / owner
continue
st = _int(f[1])
shares.append({
"id": _int(f[0]),
"share_type": self.SHARE_TYPES.get(st, str(st)),
"share_with": f[2],
"owner": f[3],
"target": f[4],
"permissions": _int(f[5]),
"created": _epoch(f[6]),
"path": f[7],
})
return shares
# -- server / apps / config ------------------------------------------------
def maintenance_mode(self, on: bool) -> str:
return self._occ_ok("maintenance:mode " + ("--on" if on else "--off")).strip()
def maintenance_status(self) -> bool:
rc, out, err = self.occ("config:system:get maintenance")
return rc == 0 and out.strip().lower() in ("true", "1", "yes")
def list_apps(self) -> Any:
return json.loads(self._occ_ok("app:list --output=json") or "{}")
def enable_app(self, app: str) -> str:
return self._occ_ok(f"app:enable {shlex.quote(app)}").strip()
def disable_app(self, app: str) -> str:
return self._occ_ok(f"app:disable {shlex.quote(app)}").strip()
def config_get(self, name: str) -> Optional[str]:
rc, out, err = self.occ(f"config:system:get {shlex.quote(name)}")
return out.strip() if rc == 0 else None
def config_list(self) -> Any:
return json.loads(self._occ_ok("config:list system --output=json") or "{}")
def config_set(self, name: str, value: str, vtype: str = "string") -> str:
return self._occ_ok(
f"config:system:set {shlex.quote(name)} "
f"--value={shlex.quote(value)} --type={shlex.quote(vtype)}").strip()
# -- files admin -----------------------------------------------------------
def files_scan(self, uid: Optional[str] = None, all_users: bool = False,
path: Optional[str] = None) -> str:
if all_users:
args = "files:scan --all"
elif uid:
args = f"files:scan {shlex.quote(uid)}"
else:
raise bc.BackupError("files_scan requires a uid or all_users=True")
if path:
args += f" --path={shlex.quote(path)}"
return self._occ_ok(args, timeout=DU_TIMEOUT).strip()
def transfer_ownership(self, src: str, dst: str,
path: Optional[str] = None) -> str:
args = (f"files:transfer-ownership {shlex.quote(src)} "
f"{shlex.quote(dst)} -s")
if path:
args += f" --path={shlex.quote(path)}"
return self._occ_ok(args, timeout=DU_TIMEOUT).strip()
def trashbin_cleanup(self, users: list[str]) -> str:
us = " ".join(shlex.quote(u) for u in users)
return self._occ_ok(f"trashbin:cleanup {us}".strip(),
timeout=DU_TIMEOUT).strip()
def versions_cleanup(self, users: list[str]) -> str:
us = " ".join(shlex.quote(u) for u in users)
return self._occ_ok(f"versions:cleanup {us}".strip(),
timeout=DU_TIMEOUT).strip()
def _int(v: Any) -> Optional[int]:
try:
return int(str(v).strip())
except (TypeError, ValueError):
return None
def _epoch(v: Any) -> Optional[str]:
"""occ emits lastLogin/creationTime as Unix epoch seconds; 0 = never."""