Laborator 5: inspectează procesele și demonstrează permisiunile Linux

180–240 de minute · Ubuntu/Kubuntu · Nivel: începător · Continuă proiectul din Laboratoarele 1–4

Ce construim și ce trebuie să înțelegi

În teoria Lecției 5 ai separat programul de proces și ai văzut că o acțiune a agentului devine reală numai printr-un proces care rulează cu o anumită identitate. În laborator transformăm ideea într-o demonstrație pe care o poți verifica singur: pornești un proces de test, îi inspectezi PID-ul, PPID-ul, UID-urile, GID-urile, grupurile suplimentare, umask, folderul curent și descriptorii de fișiere, apoi rulezi șase operații într-un mediu de test izolat în proiect.

Laboratorul continuă metoda folosită până acum: observăm înainte să modificăm, separăm rezultatele locale de artefactele versionate, scriem teste, generăm o diagramă, păstrăm un ADR și facem un commit Git numai după verificare. Nu instalăm un LLM și nu conectăm încă un agent la shell (interpretorul de comenzi). Mai întâi înțelegem ce autoritate primește procesul care ar executa acțiunile agentului.

1. Vezi de la început rezultatul final

Ca în Laboratoarele 3 și 4, începem cu harta artefactelor pentru ca fiecare comandă să aibă un loc și un scop.

ai-agents-zero-to-expert/
├── configs/
│   └── lesson05_authority_policy.json
├── data/
│   └── lesson05/
│       ├── permission_sandbox/          # generat local, ignorat de Git
│       ├── process_snapshot.json        # generat local, ignorat
│       └── permission_probe.json        # generat local, ignorat
├── diagrams/
│   ├── lesson05_authority_flow.dot
│   └── lesson05_authority_flow.png
├── docs/
│   ├── architecture-decisions/
│   │   └── ADR-0005-least-privilege-process-identity.md
│   └── lessons/
│       └── lesson-05/
│           ├── authority-design.md
│           ├── process_report.md        # generat local, ignorat
│           └── permission_probe_report.md # generat local, ignorat
├── logs/
│   └── lesson05/
│       ├── .gitkeep
│       ├── worker.pid                   # generat
│       └── worker.out                   # generat
├── src/
│   └── lesson05_permissions_lab.py
└── tests/
    └── test_lesson05_permissions.py

Separarea este intenționată. Politica, codul, testele, ADR-ul și fișa de proiectare sunt artefacte reutilizabile. PID-ul, UID-urile, căile deschise și rezultatele efective ale unui proces sunt specifice calculatorului și sesiunii; nu le păstrăm ca documentație valabilă pentru toate calculatoarele.

2. Stabilește limitele laboratorului

3. Instrumentele folosite

Laboratorul introduce aceste comenzi numai ca instrumente punctuale de observație. Lecția 6 va preda terminalul ca mediu de lucru: căi, pipe-uri, variabile de mediu, servicii, SSH și loguri.

4. Continuă proiectul cumulativ

cd ~/ai-agents-zero-to-expert
pwd

Dacă începi direct cu Laboratorul 5, creează rădăcina proiectului:

mkdir -p ~/ai-agents-zero-to-expert
cd ~/ai-agents-zero-to-expert

Creează folderele necesare. mkdir -p nu strică folderele existente:

mkdir -p \
  src \
  tests \
  configs \
  data/lesson05 \
  diagrams \
  docs/architecture-decisions \
  docs/lessons/lesson-05 \
  logs/lesson05

touch logs/lesson05/.gitkeep

5. Verifică instrumentele de bază

sudo apt update
sudo apt install -y \
  python3 \
  python3-venv \
  git \
  jq \
  tree \
  graphviz \
  acl \
  procps \
  util-linux

python3 --version
git --version
jq --version
dot -V
getfacl --version
ps --version | head -n 1
namei --version | head -n 1

[ -d .git ] || git init -b main

git config user.name >/dev/null || git config user.name "Cursant AI Agents"
git config user.email >/dev/null || git config user.email "student@local.invalid"

sudo apare aici numai pentru instalarea pachetelor de sistem. Nu folosi sudo pentru programul laboratorului.

Comanda git init inițializează Git numai dacă proiectul nu este deja un depozit Git. Următoarele două comenzi verifică dacă Git găsește deja un nume și o adresă de e-mail configurate, local sau global. Numai dacă lipsesc, setează valori locale pentru acest proiect. Dacă ai parcurs laboratoarele anterioare și identitatea există deja, comenzile nu o schimbă. Valorile Cursant AI Agents și student@local.invalid sunt potrivite pentru acest proiect didactic; într-un proiect real folosește numele și adresa ta de e-mail.

6. Creează sau activează mediul virtual Python

if [ ! -d .venv ]; then
  python3 -m venv .venv
fi

source .venv/bin/activate
which python
python --version

Programul folosește numai biblioteca standard Python; mediul virtual păstrează însă continuitatea proiectului și pregătește Lecțiile următoare.

7. Extinde .gitignore fără să publici capturi locale ale proceselor

Dacă nu există încă un .gitignore, creează baza minimă:

if [ ! -s .gitignore ]; then
  cat > .gitignore <<'EOF'
.venv/
__pycache__/
*.py[cod]
.pytest_cache/
.env
.env.*
logs/*
!logs/.gitkeep
*.tmp
.DS_Store
EOF
fi

Adaugă regulile Lecției 5 o singură dată:

if ! grep -q '^# Laboratorul 5:' .gitignore; then
  cat >> .gitignore <<'EOF'

# Laboratorul 5: sandbox și rapoarte locale/specifice procesului
/data/lesson05/permission_sandbox/
/data/lesson05/process_snapshot.json
/data/lesson05/permission_probe.json
/docs/lessons/lesson-05/process_report.md
/docs/lessons/lesson-05/permission_probe_report.md
!/logs/lesson05/
/logs/lesson05/*
!/logs/lesson05/.gitkeep
EOF
fi

touch logs/lesson05/.gitkeep

Motivul este diferit de cel din Lecția 4. Acolo catalogul era o sursă versionată și trebuia păstrat. Aici capturile de stare pot conține PID-uri, UID-uri/GID-uri, căi și descriptori care descriu numai calculatorul și sesiunea curentă.

8. Creează politica didactică de autoritate

Înainte să creăm fișierele, declarăm rezultatul așteptat. Astfel nu interpretăm după fapt dacă o operație „pare” corectă.

Creează configs/lesson05_authority_policy.json. Folosim un here-document, aceeași metodă din laboratoarele anterioare: shell-ul, adică interpretorul de comenzi, scrie în fișier tot ce se află între prima linie și marcajul final JSON.

cat > configs/lesson05_authority_policy.json <<'JSON'
{
  "schema_version": 1,
  "purpose": "Politică didactică pentru un agent care citește surse, scrie drafturi și nu primește privilegii administrative.",
  "fixture": {
    "source_dir": {
      "path": "source",
      "mode": "0555"
    },
    "source_file": {
      "path": "source/contract-a.txt",
      "mode": "0444"
    },
    "drafts_dir": {
      "path": "drafts",
      "mode": "0700"
    },
    "locked_dir": {
      "path": "locked",
      "mode": "0000"
    },
    "locked_file": {
      "path": "locked/secret.txt",
      "mode": "0600"
    },
    "deletion_dir": {
      "path": "deletion-demo",
      "mode": "0700"
    },
    "deletion_file": {
      "path": "deletion-demo/read-only-note.txt",
      "mode": "0444"
    }
  },
  "expected_operations": [
    {
      "id": "read_source",
      "description": "citește documentul-sursă",
      "expected": "allowed"
    },
    {
      "id": "append_source",
      "description": "încearcă să modifice documentul-sursă",
      "expected": "denied"
    },
    {
      "id": "create_in_source",
      "description": "încearcă să creeze un fișier nou în folderul-sursă",
      "expected": "denied"
    },
    {
      "id": "write_draft",
      "description": "creează un draft în folderul de ieșire",
      "expected": "allowed"
    },
    {
      "id": "read_locked",
      "description": "încearcă să citească printr-un folder fără drept de traversare",
      "expected": "denied"
    },
    {
      "id": "delete_readonly_in_writable_dir",
      "description": "șterge un fișier doar pentru citire dintr-un folder pe care îl poate modifica",
      "expected": "allowed"
    }
  ]
}
JSON

Verifică sintaxa și numărul de operații:

jq empty configs/lesson05_authority_policy.json
jq '.expected_operations | length' configs/lesson05_authority_policy.json

Trebuie să vezi 6. Câmpul schema_version este verificat de program și trebuie să rămână 1 pentru această versiune a laboratorului. Observă apoi cazul intenționat contraintuitiv: delete_readonly_in_writable_dir este așteptat ca allowed.

9. Creează instrumentul de laborator

Programul are trei roluri distincte: pregătește mediul controlat de test, explică modelul rwx și inspectează sau testează procese reale. Nu folosește un LLM și nu execută comenzi arbitrare primite de la utilizator.

Creează src/lesson05_permissions_lab.py direct din terminal. Marcajul PY este doar delimitatorul here-document-ului; nu face parte din fișierul Python.

cat > src/lesson05_permissions_lab.py <<'PY'
#!/usr/bin/env python3
"""Laboratorul 5 — procese, identitate și permisiuni Linux.

Instrument didactic fără dependențe externe. Lucrează numai cu:
- un sandbox explicit creat în data/lesson05/permission_sandbox;
- un proces ales explicit prin PID;
- informații locale din /proc.

Nu escaladează privilegii și nu modifică permisiuni în afara sandbox-ului.
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import stat
import sys
import time
from pathlib import Path
from typing import Any


def load_json(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise ValueError(f"Fișierul nu există: {path}") from exc
    except json.JSONDecodeError as exc:
        raise ValueError(
            f"JSON invalid în {path}: verifică sintaxa la linia {exc.lineno}, "
            f"coloana {exc.colno}"
        ) from exc
    if not isinstance(value, dict):
        raise ValueError(f"{path} trebuie să conțină un obiect JSON")
    return value


def parse_octal_mode(value: str) -> int:
    if not isinstance(value, str) or not value:
        raise ValueError("modul trebuie să fie text octal, de exemplu 0755")
    if not all(ch in "01234567" for ch in value):
        raise ValueError(f"mod octal invalid: {value!r}")
    mode = int(value, 8)
    if mode < 0 or mode > 0o7777:
        raise ValueError(f"mod în afara intervalului: {value!r}")
    return mode


def mode_triplets(mode: int) -> str:
    return stat.filemode(mode)[1:10]


REQUIRED_FIXTURE_KEYS = {
    "source_dir",
    "source_file",
    "drafts_dir",
    "locked_dir",
    "locked_file",
    "deletion_dir",
    "deletion_file",
}


def validate_policy(policy: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    if policy.get("schema_version") != 1:
        errors.append("schema_version trebuie să fie 1")

    fixture = policy.get("fixture")
    operations = policy.get("expected_operations")
    if not isinstance(fixture, dict) or not fixture:
        errors.append("fixture trebuie să fie un obiect JSON nevid")
    else:
        missing_fixture = sorted(REQUIRED_FIXTURE_KEYS - set(fixture))
        if missing_fixture:
            errors.append(
                f"intrări fixture lipsă: {', '.join(missing_fixture)}"
            )
        for name, item in fixture.items():
            if not isinstance(item, dict):
                errors.append(f"fixture.{name} trebuie să fie obiect")
                continue
            path_value = item.get("path")
            if not isinstance(path_value, str) or not path_value:
                errors.append(f"fixture.{name}.path lipsește")
            elif not _is_safe_relative_path(path_value):
                errors.append(
                    f"fixture.{name}.path trebuie să fie o cale relativă din sandbox, fără '..'"
                )
            try:
                parse_octal_mode(item.get("mode", ""))
            except ValueError as exc:
                errors.append(f"fixture.{name}.mode: {exc}")
    if not isinstance(operations, list) or not operations:
        errors.append("expected_operations trebuie să fie o listă nevidă")
    else:
        seen: set[str] = set()
        for item in operations:
            if not isinstance(item, dict):
                errors.append("fiecare operație trebuie să fie obiect")
                continue
            op_id = item.get("id")
            if not isinstance(op_id, str) or not op_id:
                errors.append("operație fără id")
            elif op_id in seen:
                errors.append(f"id duplicat: {op_id}")
            else:
                seen.add(op_id)
            if item.get("expected") not in {"allowed", "denied"}:
                errors.append(f"{op_id or '?'}: expected trebuie să fie allowed sau denied")

        if all(isinstance(item, dict) for item in operations):
            ids = {
                item.get("id")
                for item in operations
                if isinstance(item.get("id"), str) and item.get("id")
            }
            unknown = sorted(ids - SUPPORTED_OPERATIONS)
            missing = sorted(SUPPORTED_OPERATIONS - ids)
            if unknown:
                errors.append(f"operații necunoscute: {', '.join(unknown)}")
            if missing:
                errors.append(f"operații lipsă: {', '.join(missing)}")
    return errors


def _safe_sandbox_root(root: Path) -> Path:
    project_root = Path.cwd().resolve()
    expected = project_root / "data" / "lesson05" / "permission_sandbox"

    raw = root.expanduser()
    if raw.is_absolute():
        candidate = Path(os.path.abspath(raw))
    else:
        candidate = Path(os.path.abspath(project_root / raw))

    if candidate != expected:
        raise ValueError(
            "Refuz să modific această cale. Sandbox-ul trebuie să fie exact "
            "data/lesson05/permission_sandbox din proiectul curent."
        )

    for component in (
        project_root / "data",
        project_root / "data" / "lesson05",
        expected,
    ):
        if component.exists() and component.is_symlink():
            raise ValueError(
                f"Refuz sandbox-ul: {component} este o legătură simbolică."
            )

    return expected


def _relax_for_cleanup(root: Path) -> None:
    if not root.exists():
        return
    for current, dirs, files in os.walk(root, topdown=True):
        current_path = Path(current)
        try:
            current_path.chmod(0o700)
        except OSError:
            pass
        for name in dirs:
            try:
                (current_path / name).chmod(0o700)
            except OSError:
                pass
        for name in files:
            try:
                (current_path / name).chmod(0o600)
            except OSError:
                pass


def setup_fixture(policy: dict[str, Any], root: Path) -> None:
    errors = validate_policy(policy)
    if errors:
        raise ValueError("; ".join(errors))
    root = _safe_sandbox_root(root)
    if root.exists():
        _relax_for_cleanup(root)
        shutil.rmtree(root)
    root.mkdir(parents=True)

    source = root / policy["fixture"]["source_dir"]["path"]
    drafts = root / policy["fixture"]["drafts_dir"]["path"]
    locked = root / policy["fixture"]["locked_dir"]["path"]
    deletion = root / policy["fixture"]["deletion_dir"]["path"]
    for path in (source, drafts, locked, deletion):
        path.mkdir(parents=True, exist_ok=True)

    (root / policy["fixture"]["source_file"]["path"]).write_text(
        "Contract fictiv pentru laboratorul 5.\n", encoding="utf-8"
    )
    (root / policy["fixture"]["locked_file"]["path"]).write_text(
        "Secret fictiv. Nu este o credențială reală.\n", encoding="utf-8"
    )
    (root / policy["fixture"]["deletion_file"]["path"]).write_text(
        "Acest fișier este doar pentru citire, dar folderul părinte permite modificări.\n",
        encoding="utf-8",
    )

    for name in ("source_file", "locked_file", "deletion_file"):
        item = policy["fixture"][name]
        (root / item["path"]).chmod(parse_octal_mode(item["mode"]))
    for name in ("source_dir", "drafts_dir", "deletion_dir", "locked_dir"):
        item = policy["fixture"][name]
        (root / item["path"]).chmod(parse_octal_mode(item["mode"]))


def select_permission_class(
    object_uid: int,
    object_gid: int,
    effective_uid: int,
    effective_gid: int,
    supplementary_groups: set[int],
) -> str:
    if effective_uid == object_uid:
        return "owner"
    if object_gid == effective_gid or object_gid in supplementary_groups:
        return "group"
    return "others"


def class_bits(mode: int, permission_class: str) -> int:
    perm = stat.S_IMODE(mode)
    if permission_class == "owner":
        return (perm >> 6) & 0b111
    if permission_class == "group":
        return (perm >> 3) & 0b111
    if permission_class == "others":
        return perm & 0b111
    raise ValueError(f"clasă necunoscută: {permission_class}")


ACTION_BIT = {"read": 0b100, "write": 0b010, "execute": 0b001, "traverse": 0b001}

SUPPORTED_OPERATIONS = {
    "read_source",
    "append_source",
    "create_in_source",
    "write_draft",
    "read_locked",
    "delete_readonly_in_writable_dir",
}


def _is_safe_relative_path(value: str) -> bool:
    path = Path(value)
    return (
        bool(value)
        and not path.is_absolute()
        and ".." not in path.parts
        and path != Path(".")
    )


def basic_mode_allows(
    mode: int,
    object_uid: int,
    object_gid: int,
    action: str,
    effective_uid: int,
    effective_gid: int,
    supplementary_groups: set[int],
) -> tuple[bool, str]:
    """Evaluează numai biții clasici proprietar/grup/ceilalți.

    Pentru simplitate didactică folosește EUID/EGID și grupurile suplimentare.
    Linux folosește FSUID/FSGID la verificările de acces la fișiere; în mod
    obișnuit acestea urmează EUID/EGID. Funcția nu modelează ACL-uri,
    capabilities, LSM, opțiuni de montare sau tratamentul special al UID 0.
    Este intenționat un explicator pentru modelul de bază, nu un simulator de kernel.
    """
    if action not in ACTION_BIT:
        raise ValueError(f"acțiune necunoscută: {action}")
    permission_class = select_permission_class(
        object_uid, object_gid, effective_uid, effective_gid, supplementary_groups
    )
    bits = class_bits(mode, permission_class)
    return bool(bits & ACTION_BIT[action]), permission_class


def explain_path(path: Path, action: str) -> dict[str, Any]:
    try:
        info = path.stat()
    except FileNotFoundError as exc:
        raise ValueError(f"Calea nu există: {path}") from exc
    except PermissionError as exc:
        raise ValueError(f"Nu pot inspecta calea din cauza permisiunilor: {path}") from exc

    euid = os.geteuid()
    egid = os.getegid()
    groups = set(os.getgroups())
    allowed, permission_class = basic_mode_allows(
        info.st_mode, info.st_uid, info.st_gid, action, euid, egid, groups
    )
    return {
        "path": str(path),
        "type": "folder" if path.is_dir() else "fișier",
        "mode_octal": f"{stat.S_IMODE(info.st_mode):04o}",
        "mode_symbolic": mode_triplets(info.st_mode),
        "owner_uid": info.st_uid,
        "group_gid": info.st_gid,
        "effective_uid": euid,
        "effective_gid": egid,
        "supplementary_groups": sorted(groups),
        "selected_class": permission_class,
        "action": action,
        "basic_mode_allows": allowed,
        "warning": (
            "Acesta este numai modelul clasic rwx. Explicația folosește EUID/EGID "
            "ca aproximație didactică pentru FSUID/FSGID. Verdictul kernelului poate fi "
            "influențat de ACL, capabilities, LSM, opțiuni de montare și alte reguli."
        ),
    }


def explain_delete(path: Path) -> dict[str, Any]:
    parent = path.parent
    write = explain_path(parent, "write")
    traverse = explain_path(parent, "traverse")
    return {
        "path": str(path),
        "parent": str(parent),
        "parent_write": write["basic_mode_allows"],
        "parent_traverse": traverse["basic_mode_allows"],
        "basic_mode_allows_delete": bool(
            write["basic_mode_allows"] and traverse["basic_mode_allows"]
        ),
        "note": (
            "Pentru ștergerea unei intrări, biții de bază relevanți sunt cei ai "
            "folderului părinte; sticky bit și alte politici pot adăuga restricții."
        ),
    }


def parse_proc_status(text: str) -> dict[str, Any]:
    wanted = {"Name", "State", "Pid", "PPid", "Uid", "Gid", "Groups", "Umask"}
    result: dict[str, Any] = {}
    for line in text.splitlines():
        if ":" not in line:
            continue
        key, raw = line.split(":", 1)
        if key not in wanted:
            continue
        value = raw.strip()
        if key in {"Pid", "PPid"}:
            result[key.lower()] = int(value)
        elif key in {"Uid", "Gid", "Groups"}:
            result[key.lower()] = [int(item) for item in value.split()]
        elif key == "Umask":
            result["umask"] = value
        elif key == "Name":
            result["name"] = value
        elif key == "State":
            result["state"] = value
    return result


def redact_path(value: str, project_root: Path | None) -> str:
    if project_root is None:
        return value
    root = str(project_root.expanduser().resolve())
    return value.replace(root, "<project-root>")


def inspect_process(pid: int, project_root: Path | None = None) -> dict[str, Any]:
    proc = Path("/proc") / str(pid)
    if not proc.exists():
        raise ValueError(f"PID-ul {pid} nu există sau nu mai este vizibil în /proc")
    status_path = proc / "status"
    try:
        status = parse_proc_status(status_path.read_text(encoding="utf-8"))
    except OSError as exc:
        raise ValueError(f"Nu pot citi {status_path}: {exc}") from exc

    def safe_link(path: Path) -> str:
        try:
            return redact_path(os.readlink(path), project_root)
        except OSError as exc:
            return f"<indisponibil: {exc.strerror or exc}>"

    try:
        cmd_bytes = (proc / "cmdline").read_bytes()
        cmdline = " ".join(
            part.decode("utf-8", "replace")
            for part in cmd_bytes.split(b"\0")
            if part
        )
        cmdline = redact_path(cmdline, project_root)
    except OSError as exc:
        cmdline = f"<indisponibil: {exc.strerror or exc}>"

    fds: list[dict[str, Any]] = []
    try:
        for entry in sorted((proc / "fd").iterdir(), key=lambda p: int(p.name)):
            try:
                target = redact_path(os.readlink(entry), project_root)
            except OSError as exc:
                target = f"<indisponibil: {exc.strerror or exc}>"
            fds.append({"fd": int(entry.name), "target": target})
    except OSError:
        pass

    return {
        **status,
        "cwd": safe_link(proc / "cwd"),
        "exe": safe_link(proc / "exe"),
        "cmdline": cmdline,
        "open_file_descriptors": fds,
        "privacy_note": (
            "Raport local: programul nu adaugă explicit hostname-ul sau numele de utilizator. "
            "Căile din project_root sunt înlocuite cu <project-root>, dar linia de comandă și "
            "alte căi pot conține în continuare informații locale. Fișierul rămâne ignorat de Git."
        ),
    }


def process_report_markdown(snapshot: dict[str, Any]) -> str:
    lines = [
        "# Raport local de proces — Lecția 5",
        "",
        f"- Nume proces: `{snapshot.get('name', '?')}`",
        f"- PID: `{snapshot.get('pid', '?')}`",
        f"- PPID: `{snapshot.get('ppid', '?')}`",
        f"- Stare: `{snapshot.get('state', '?')}`",
        f"- UID-uri (real / efectiv / salvat / sistem de fișiere): `{snapshot.get('uid', [])}`",
        f"- GID-uri (real / efectiv / salvat / sistem de fișiere): `{snapshot.get('gid', [])}`",
        f"- Grupuri suplimentare: `{snapshot.get('groups', [])}`",
        f"- Umask: `{snapshot.get('umask', '?')}`",
        f"- Folder curent (CWD): `{snapshot.get('cwd', '?')}`",
        f"- Executabil: `{snapshot.get('exe', '?')}`",
        f"- Linie de comandă: `{snapshot.get('cmdline', '?')}`",
        "",
        "## Descriptori de fișiere deschiși",
        "",
    ]
    fds = snapshot.get("open_file_descriptors", [])
    if fds:
        for item in fds:
            lines.append(f"- FD `{item['fd']}` → `{item['target']}`")
    else:
        lines.append("- Niciun descriptor nu a putut fi enumerat.")
    lines += [
        "",
        "## Interpretare",
        "",
        "Procesul execută acțiuni cu identitatea numerică de mai sus. "
        "PID-ul îl identifică pe durata acestei execuții; nu este identitatea programului în general.",
        "",
        f"> {snapshot.get('privacy_note', '')}",
        "",
    ]
    return "\n".join(lines)


def _attempt(fn) -> tuple[str, str]:
    try:
        fn()
        return "allowed", "operația a reușit"
    except PermissionError as exc:
        return "denied", f"PermissionError: {exc.strerror or exc}"
    except OSError as exc:
        return "error", f"{type(exc).__name__}: {exc}"


def run_permission_probe(policy: dict[str, Any], root: Path) -> dict[str, Any]:
    if os.geteuid() == 0:
        raise ValueError(
            "Refuz să rulez proba ca root. Rulează fără sudo; UID 0 ar ocoli "
            "o parte din limitele pe care laboratorul încearcă să le demonstreze."
        )
    root = _safe_sandbox_root(root)
    if not root.is_dir():
        raise ValueError("Sandbox-ul nu există. Rulează mai întâi subcomanda setup.")
    source_file = root / policy["fixture"]["source_file"]["path"]
    source_dir = root / policy["fixture"]["source_dir"]["path"]
    drafts_dir = root / policy["fixture"]["drafts_dir"]["path"]
    locked_file = root / policy["fixture"]["locked_file"]["path"]
    deletion_file = root / policy["fixture"]["deletion_file"]["path"]

    def read_source():
        source_file.read_text(encoding="utf-8")

    def append_source():
        with source_file.open("a", encoding="utf-8") as handle:
            handle.write("NU AR TREBUI SĂ APARĂ\n")

    def create_in_source():
        candidate = source_dir / "should-not-exist.txt"
        candidate.write_text("test\n", encoding="utf-8")
        candidate.unlink(missing_ok=True)

    def write_draft():
        candidate = drafts_dir / "analysis-draft.txt"
        candidate.write_text("Draft generat în laboratorul 5.\n", encoding="utf-8")

    def read_locked():
        locked_file.read_text(encoding="utf-8")

    def delete_readonly():
        deletion_file.unlink()
        deletion_file.write_text(
            "Fișier doar pentru citire recreat după demonstrația de ștergere.\n",
            encoding="utf-8",
        )
        deletion_file.chmod(0o444)

    attempts = {
        "read_source": read_source,
        "append_source": append_source,
        "create_in_source": create_in_source,
        "write_draft": write_draft,
        "read_locked": read_locked,
        "delete_readonly_in_writable_dir": delete_readonly,
    }
    expected = {item["id"]: item["expected"] for item in policy["expected_operations"]}
    results = []
    for op_id, fn in attempts.items():
        observed, detail = _attempt(fn)
        results.append({
            "id": op_id,
            "expected": expected[op_id],
            "observed": observed,
            "matches_expected": observed == expected[op_id],
            "detail": detail,
        })
    return {
        "effective_uid": os.geteuid(),
        "effective_gid": os.getegid(),
        "sandbox": str(root),
        "results": results,
        "all_expected": all(item["matches_expected"] for item in results),
    }


def probe_report_markdown(result: dict[str, Any]) -> str:
    status_ro = {
        "allowed": "permis",
        "denied": "refuzat",
        "error": "eroare",
    }
    lines = [
        "# Raport al probelor de permisiuni — Lecția 5",
        "",
        f"- EUID: `{result['effective_uid']}`",
        f"- EGID: `{result['effective_gid']}`",
        f"- Sandbox: `{result['sandbox']}`",
        f"- Toate rezultatele corespund politicii: **{'DA' if result['all_expected'] else 'NU'}**",
        "",
        "| Operație | Așteptat | Observat | Potrivire |",
        "|---|---|---|---|",
    ]
    for item in result["results"]:
        lines.append(
            f"| `{item['id']}` | {status_ro.get(item['expected'], item['expected'])} | "
            f"{status_ro.get(item['observed'], item['observed'])} | "
            f"{'✓' if item['matches_expected'] else '✗'} |"
        )
    lines += [
        "",
        "## Lecția importantă",
        "",
        "Cazul `delete_readonly_in_writable_dir` arată că ștergerea unei intrări "
        "este în primul rând o operație asupra folderului părinte. Fișierul "
        "poate fi doar pentru citire și totuși șters dacă regulile folderului permit operația.",
        "",
    ]
    return "\n".join(lines)


def authority_flow_dot() -> str:
    return r"""digraph authority_flow {
  rankdir=LR;
  graph [bgcolor="transparent", pad="0.2", nodesep="0.35", ranksep="0.55"];
  node [fontname="DejaVu Sans", fontsize=10, shape=box, style="rounded"];
  edge [fontname="DejaVu Sans", fontsize=9];

  agent [label="Agent / controller\nalege acțiunea"];
  tool [label="Instrument\ntransformă decizia în operație"];
  process [label="Proces Linux\nPID + stare"];
  identity [label="Identitate proces\nEUID / EGID / grupuri"];
  kernel [label="Kernel\nverifică accesul", color="red", penwidth=2];
  object [label="Obiect\nfișier / folder / proces"];
  allow [label="PERMIS", color="green", fontcolor="green"];
  deny [label="REFUZAT", color="red", fontcolor="red"];

  agent -> tool;
  tool -> process;
  process -> identity;
  identity -> kernel;
  object -> kernel [label="proprietar/grup/rwx\n+ reguli suplimentare"];
  kernel -> allow;
  kernel -> deny;
}
"""


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(value, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def write_text(path: Path, value: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(value, encoding="utf-8")


def hold_worker(root: Path, pid_file: Path, seconds: int) -> int:
    if seconds <= 0:
        raise ValueError("--seconds trebuie să fie un număr pozitiv")
    root = _safe_sandbox_root(root)
    source_file = root / "source" / "contract-a.txt"
    draft_file = root / "drafts" / "worker-heartbeat.txt"
    if not source_file.exists():
        raise ValueError("Sandbox-ul nu există. Rulează mai întâi subcomanda setup.")

    with source_file.open("r", encoding="utf-8") as source_handle, draft_file.open(
        "a", encoding="utf-8"
    ) as draft_handle:
        draft_handle.write(f"proces de test PID={os.getpid()} pornit\n")
        draft_handle.flush()
        _ = source_handle.readline()

        pid_file.parent.mkdir(parents=True, exist_ok=True)
        pid_file.write_text(f"{os.getpid()}\n", encoding="utf-8")
        time.sleep(seconds)
    return 0


def _require_unprivileged(command: str) -> None:
    if os.geteuid() == 0 and command in {
        "setup", "explain", "explain-delete", "hold", "probe"
    }:
        raise ValueError(
            f"Refuz să rulez subcomanda {command!r} ca root. "
            "Rulează programul laboratorului fără sudo."
        )


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Laboratorul 5: inspectează procese și explică permisiuni Linux."
    )
    parser.add_argument(
        "--policy",
        type=Path,
        default=Path("configs/lesson05_authority_policy.json"),
    )
    sub = parser.add_subparsers(dest="command", required=True)

    setup = sub.add_parser("setup", help="Creează mediul didactic de test.")
    setup.add_argument(
        "--sandbox",
        type=Path,
        default=Path("data/lesson05/permission_sandbox"),
    )

    explain = sub.add_parser("explain", help="Explică biții rwx de bază pentru o cale.")
    explain.add_argument("path", type=Path)
    explain.add_argument("action", choices=["read", "write", "execute", "traverse"])

    delete = sub.add_parser(
        "explain-delete",
        help="Explică ștergerea prin folderul părinte.",
    )
    delete.add_argument("path", type=Path)

    hold = sub.add_parser("hold", help="Pornește un proces de test simplu, ușor de inspectat.")
    hold.add_argument(
        "--sandbox",
        type=Path,
        default=Path("data/lesson05/permission_sandbox"),
    )
    hold.add_argument(
        "--pid-file",
        type=Path,
        default=Path("logs/lesson05/worker.pid"),
    )
    hold.add_argument("--seconds", type=int, default=180)

    inspect = sub.add_parser("inspect", help="Inspectează un PID prin /proc.")
    inspect.add_argument("pid", type=int)
    inspect.add_argument("--project-root", type=Path, default=Path.cwd())
    inspect.add_argument("--json-output", type=Path)
    inspect.add_argument("--report", type=Path)

    probe = sub.add_parser("probe", help="Rulează probe sigure numai în mediul de test.")
    probe.add_argument(
        "--sandbox",
        type=Path,
        default=Path("data/lesson05/permission_sandbox"),
    )
    probe.add_argument("--json-output", type=Path)
    probe.add_argument("--report", type=Path)

    dot = sub.add_parser("dot", help="Generează diagrama conceptuală DOT.")
    dot.add_argument("--output", type=Path, required=True)

    args = parser.parse_args()

    try:
        policy = load_json(args.policy)
        errors = validate_policy(policy)
        if errors:
            for error in errors:
                print(f"EROARE POLITICĂ: {error}", file=sys.stderr)
            return 2

        _require_unprivileged(args.command)

        if args.command == "setup":
            setup_fixture(policy, args.sandbox)
            print(f"Mediu de test creat: {args.sandbox}")
            return 0

        if args.command == "explain":
            print(
                json.dumps(
                    explain_path(args.path, args.action),
                    ensure_ascii=False,
                    indent=2,
                )
            )
            return 0

        if args.command == "explain-delete":
            print(
                json.dumps(
                    explain_delete(args.path),
                    ensure_ascii=False,
                    indent=2,
                )
            )
            return 0

        if args.command == "hold":
            return hold_worker(args.sandbox, args.pid_file, args.seconds)

        if args.command == "inspect":
            snapshot = inspect_process(args.pid, args.project_root)
            if args.json_output:
                write_json(args.json_output, snapshot)
            report = process_report_markdown(snapshot)
            if args.report:
                write_text(args.report, report)
            print(report, end="")
            return 0

        if args.command == "probe":
            result = run_permission_probe(policy, args.sandbox)
            if args.json_output:
                write_json(args.json_output, result)
            report = probe_report_markdown(result)
            if args.report:
                write_text(args.report, report)
            print(report, end="")
            return 0 if result["all_expected"] else 1

        if args.command == "dot":
            write_text(args.output, authority_flow_dot())
            print(args.output)
            return 0

    except (ValueError, OSError) as exc:
        print(f"EROARE: {exc}", file=sys.stderr)
        return 2

    return 2


if __name__ == "__main__":
    raise SystemExit(main())
PY

Verifică sintaxa înainte de orice probă:

python -m py_compile src/lesson05_permissions_lab.py

10. Creează sandbox-ul de permisiuni

python src/lesson05_permissions_lab.py setup

Structura creată este:

data/lesson05/permission_sandbox/
├── source/
│   └── contract-a.txt
├── drafts/
├── locked/
│   └── secret.txt
└── deletion-demo/
    └── read-only-note.txt

Pentru comanda setup, programul acceptă numai calea exactă data/lesson05/permission_sandbox din proiectul curent. În plus, refuză componente simbolice pentru această cale și respinge căile din politică ce încearcă să iasă din sandbox prin ... Protecțiile există tocmai pentru a evita ca un argument greșit să ducă la modificarea altor foldere.

11. Inspectează permisiunile înainte să testezi accesul

Mai întâi citim dovezile. Abia după aceea încercăm operațiile.

stat -c '%A  %a  uid=%u gid=%g  %n' \
  data/lesson05/permission_sandbox/source \
  data/lesson05/permission_sandbox/source/contract-a.txt \
  data/lesson05/permission_sandbox/drafts \
  data/lesson05/permission_sandbox/locked \
  data/lesson05/permission_sandbox/deletion-demo \
  data/lesson05/permission_sandbox/deletion-demo/read-only-note.txt

Trebuie să recunoști aproximativ:

dr-xr-xr-x  555  …/source
-r--r--r--  444  …/source/contract-a.txt
drwx------  700  …/drafts
d---------  0    …/locked
drwx------  700  …/deletion-demo
-r--r--r--  444  …/deletion-demo/read-only-note.txt

Primul caracter arată tipul obiectului: d pentru folder și - pentru fișier obișnuit. Pentru modul 0000, stat afișează numeric un singur 0; permisiunile sunt tot zero pentru proprietar, grup și ceilalți.

Inspectează calea completă către contract:

namei -l data/lesson05/permission_sandbox/source/contract-a.txt

namei -l este util deoarece accesul la fișier depinde și de traversarea folderelor părinte.

Inspectează ACL-ul de bază:

getfacl -p data/lesson05/permission_sandbox/source/contract-a.txt
getfacl -p data/lesson05/permission_sandbox/drafts

Structura de test nu adaugă ACL-uri extinse. Folosim totuși getfacl pentru a deprinde obiceiul corect: verificăm regulile existente înainte să presupunem că vedem totul din rwx.

12. Cere programului să explice modelul rwx

Explică de ce procesul curent poate citi contractul:

python src/lesson05_permissions_lab.py explain \
  data/lesson05/permission_sandbox/source/contract-a.txt \
  read

Apoi verifică scrierea:

python src/lesson05_permissions_lab.py explain \
  data/lesson05/permission_sandbox/source/contract-a.txt \
  write

Și traversarea folderului:

python src/lesson05_permissions_lab.py explain \
  data/lesson05/permission_sandbox/source \
  traverse

Ieșirea arată UID-ul proprietarului, GID-ul grupului, EUID (UID-ul efectiv), EGID (GID-ul efectiv), grupurile suplimentare, clasa selectată și verdictul obținut din modelul clasic. Pentru verificările reale de acces la fișiere, Linux folosește FSUID și FSGID — identificatorii dedicați sistemului de fișiere. În situațiile obișnuite aceștia coincid cu EUID și EGID, dar explicatorul nostru nu încearcă să modeleze toate excepțiile kernelului.

13. Explică de ce ștergerea privește folderul părinte

Înainte să executăm operația, cere explicația:

python src/lesson05_permissions_lab.py explain-delete \
  data/lesson05/permission_sandbox/deletion-demo/read-only-note.txt

Fișierul are 0444, dar folderul părinte are 0700 și aparține procesului curent. Explicatorul trebuie să arate parent_write: true, parent_traverse: true și basic_mode_allows_delete: true.

Aceasta este una dintre cele mai importante corecții conceptuale ale lecției: „fișier doar pentru citire” nu înseamnă automat „numele nu poate fi șters”.

14. Pornește un proces controlat pe care îl poți inspecta

Nu alegem un proces de sistem aleator. Pornim un proces de test simplu care citește contractul, ține un fișier de draft deschis și apoi așteaptă.

rm -f logs/lesson05/worker.pid

python src/lesson05_permissions_lab.py hold \
  --seconds 300 \
  > logs/lesson05/worker.out 2>&1 &

for i in {1..20}; do
  [ -s logs/lesson05/worker.pid ] && break
  sleep 0.1
done

PID="$(cat logs/lesson05/worker.pid)"
echo "PID proces de test: $PID"

Ștergem mai întâi orice fișier PID rămas de la o rulare anterioară, ca să nu inspectăm din greșeală un PID vechi. Procesul de test rămâne activ aproximativ cinci minute, suficient pentru observație. Dacă expiră, pornește-l din nou.

15. Inspectează manual procesul

Începe cu o vedere compactă:

ps -o pid,ppid,user,group,state,comm,args -p "$PID"

Apoi citește câmpurile relevante direct din /proc:

grep -E '^(Name|Umask|State|Pid|PPid|Uid|Gid|Groups):'   "/proc/$PID/status"

Observă patru valori la Uid și Gid. Ele corespund identităților: reală, efectivă, salvată și cea pentru sistemul de fișiere (filesystem). Pentru acest laborator nu trebuie să le manipulezi; trebuie doar să vezi că „utilizatorul procesului” este mai precis decât un singur nume afișat.

Inspectează folderul curent și executabilul:

readlink "/proc/$PID/cwd"
readlink "/proc/$PID/exe"

Inspectează descriptorii deschiși:

ls -l "/proc/$PID/fd"

Ar trebui să apară, pe lângă intrările standard 0/1/2, descriptorul deschis pentru source/contract-a.txt și unul pentru drafts/worker-heartbeat.txt. Aceasta leagă teoria despre descriptorii de fișiere de un proces real.

16. Generează o captură structurată a stării procesului

Acum folosim instrumentul Python pentru aceeași observație:

python src/lesson05_permissions_lab.py inspect "$PID" \
  --project-root "$(pwd)" \
  --json-output data/lesson05/process_snapshot.json \
  --report docs/lessons/lesson-05/process_report.md

Programul nu adaugă explicit hostname-ul sau numele utilizatorului și înlocuiește rădăcina proiectului cu <project-root>. Totuși, linia de comandă și alte căi pot conține informații locale. De aceea raportul rămâne un artefact local, ignorat de Git, nu o documentație care trebuie publicată.

Verifică explicit:

git check-ignore -v data/lesson05/process_snapshot.json
git check-ignore -v docs/lessons/lesson-05/process_report.md

17. Rulează cele șase probe reale de permisiune

Acum trecem de la „ce credem că permit biții” la „ce a permis sau refuzat efectiv kernelul”.

Asigură-te că nu ești root:

id -u

Rezultatul trebuie să fie diferit de 0. Apoi rulează:

python src/lesson05_permissions_lab.py probe \
  --json-output data/lesson05/permission_probe.json \
  --report docs/lessons/lesson-05/permission_probe_report.md

Rezultatul așteptat este:

| Operație | Așteptat | Observat | Potrivire |
|---|---|---|---|
| `read_source` | permis | permis | ✓ |
| `append_source` | refuzat | refuzat | ✓ |
| `create_in_source` | refuzat | refuzat | ✓ |
| `write_draft` | permis | permis | ✓ |
| `read_locked` | refuzat | refuzat | ✓ |
| `delete_readonly_in_writable_dir` | permis | permis | ✓ |

În versiunea verificată a laboratorului, toate cele șase rezultate au corespuns politicii într-o execuție reală sub un utilizator Linux neprivilegiat.

18. Oprește procesul de test cu SIGTERM și recapitulează diferența față de SIGKILL

Trimite mai întâi semnalul obișnuit de terminare procesului de test:

kill -TERM "$PID"
sleep 0.2
ps -p "$PID" || true

În mod normal, procesul dispare. În acest laborator, procesul de test nu instalează un handler special pentru SIGTERM, așa că este încheiat prin comportamentul implicit al sistemului. În aplicații reale, un program poate trata SIGTERM pentru a închide fișiere, a salva stare și a opri controlat. SIGKILL este diferit: kernelul termină procesul imediat, iar procesul nu îl poate intercepta sau ignora. De aceea începem cu SIGTERM și păstrăm SIGKILL pentru situațiile în care oprirea normală nu funcționează.

Dacă procesul de test s-a încheiat singur înainte să ajungi aici, nu este o problemă. PID-ul este temporar și are sens numai cât timp procesul respectiv există; tocmai aceasta este una dintre lecțiile exercițiului.

19. Adaugă testele automate

Probele reale verifică sistemul de fișiere al calculatorului pe care rulezi. Testele automate verifică separat regulile pure pe care vrem să le păstrăm stabile.

Creează tests/test_lesson05_permissions.py. Testele nu modifică permisiunile sistemului; verifică funcțiile pure, validarea politicii și protecțiile de cale.

cat > tests/test_lesson05_permissions.py <<'PY'
from __future__ import annotations

import importlib.util
import json
import os
import stat
import unittest
from copy import deepcopy
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def load_module(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Nu pot încărca {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


lab = load_module(
    "lesson05_permissions_lab",
    ROOT / "src" / "lesson05_permissions_lab.py",
)
POLICY = json.loads(
    (ROOT / "configs" / "lesson05_authority_policy.json").read_text(
        encoding="utf-8"
    )
)


class ModeTests(unittest.TestCase):
    def test_mode_triplets_0644(self):
        mode = stat.S_IFREG | 0o644
        self.assertEqual(lab.mode_triplets(mode), "rw-r--r--")

    def test_mode_triplets_0750(self):
        mode = stat.S_IFDIR | 0o750
        self.assertEqual(lab.mode_triplets(mode), "rwxr-x---")

    def test_owner_class_has_priority(self):
        selected = lab.select_permission_class(
            1000, 2000, 1000, 2000, {2000}
        )
        self.assertEqual(selected, "owner")

    def test_supplementary_group_matches_group_class(self):
        selected = lab.select_permission_class(
            999, 2000, 1000, 3000, {2000, 4000}
        )
        self.assertEqual(selected, "group")

    def test_others_when_no_identity_matches(self):
        selected = lab.select_permission_class(
            999, 2000, 1000, 3000, {4000}
        )
        self.assertEqual(selected, "others")

    def test_owner_does_not_accumulate_group_write(self):
        allowed, selected = lab.basic_mode_allows(
            stat.S_IFREG | 0o460,
            object_uid=1000,
            object_gid=2000,
            action="write",
            effective_uid=1000,
            effective_gid=2000,
            supplementary_groups={2000},
        )
        self.assertEqual(selected, "owner")
        self.assertFalse(allowed)

    def test_file_owner_read_allowed(self):
        allowed, _ = lab.basic_mode_allows(
            stat.S_IFREG | 0o400,
            1000,
            2000,
            "read",
            1000,
            3000,
            set(),
        )
        self.assertTrue(allowed)

    def test_file_owner_write_denied(self):
        allowed, _ = lab.basic_mode_allows(
            stat.S_IFREG | 0o400,
            1000,
            2000,
            "write",
            1000,
            3000,
            set(),
        )
        self.assertFalse(allowed)

    def test_directory_x_maps_to_traverse(self):
        allowed, _ = lab.basic_mode_allows(
            stat.S_IFDIR | 0o100,
            1000,
            2000,
            "traverse",
            1000,
            3000,
            set(),
        )
        self.assertTrue(allowed)

    def test_explain_path_reports_missing_path_cleanly(self):
        with self.assertRaisesRegex(ValueError, "Calea nu există"):
            lab.explain_path(Path("/definitely/missing/lesson05"), "read")


class PolicyTests(unittest.TestCase):
    def test_policy_is_valid(self):
        self.assertEqual(lab.validate_policy(POLICY), [])

    def test_policy_rejects_unknown_schema_version(self):
        policy = deepcopy(POLICY)
        policy["schema_version"] = 2
        errors = lab.validate_policy(policy)
        self.assertIn("schema_version trebuie să fie 1", errors)

    def test_policy_has_six_expected_operations(self):
        self.assertEqual(len(POLICY["expected_operations"]), 6)

    def test_parse_octal_rejects_non_octal(self):
        with self.assertRaises(ValueError):
            lab.parse_octal_mode("08xx")

    def test_safe_sandbox_rejects_arbitrary_path(self):
        with self.assertRaises(ValueError):
            lab._safe_sandbox_root(Path("/tmp/not-the-lab"))


    def test_safe_sandbox_rejects_lookalike_path(self):
        with self.assertRaises(ValueError):
            lab._safe_sandbox_root(Path("/tmp/data/lesson05/permission_sandbox"))

    def test_policy_rejects_parent_traversal(self):
        policy = deepcopy(POLICY)
        policy["fixture"]["source_file"]["path"] = "../outside.txt"
        errors = lab.validate_policy(policy)
        self.assertTrue(any("cale relativă" in item for item in errors))

    def test_policy_rejects_missing_fixture_entry(self):
        policy = deepcopy(POLICY)
        del policy["fixture"]["locked_dir"]
        errors = lab.validate_policy(policy)
        self.assertTrue(any("intrări fixture lipsă" in item for item in errors))

    def test_policy_rejects_unknown_operation(self):
        policy = deepcopy(POLICY)
        policy["expected_operations"].append(
            {"id": "do_anything", "description": "invalid", "expected": "allowed"}
        )
        errors = lab.validate_policy(policy)
        self.assertTrue(any("operații necunoscute" in item for item in errors))

    def test_policy_rejects_missing_operation(self):
        policy = deepcopy(POLICY)
        policy["expected_operations"] = policy["expected_operations"][:-1]
        errors = lab.validate_policy(policy)
        self.assertTrue(any("operații lipsă" in item for item in errors))


class ProcTests(unittest.TestCase):
    def test_parse_proc_status(self):
        sample = """Name:\tpython
Umask:\t0022
State:\tS (sleeping)
Pid:\t4242
PPid:\t4000
Uid:\t1000\t1000\t1000\t1000
Gid:\t1000\t1000\t1000\t1000
Groups:\t4 27 1000
"""
        parsed = lab.parse_proc_status(sample)
        self.assertEqual(parsed["pid"], 4242)
        self.assertEqual(parsed["ppid"], 4000)
        self.assertEqual(parsed["uid"], [1000, 1000, 1000, 1000])
        self.assertEqual(parsed["groups"], [4, 27, 1000])
        self.assertEqual(parsed["umask"], "0022")

    def test_redact_project_root(self):
        root = Path("/home/example/project")
        value = "/home/example/project/data/file.txt"
        self.assertEqual(
            lab.redact_path(value, root),
            "<project-root>/data/file.txt",
        )

    def test_inspect_current_process_has_pid_and_uid(self):
        snapshot = lab.inspect_process(os.getpid(), ROOT)
        self.assertEqual(snapshot["pid"], os.getpid())
        self.assertTrue(snapshot["uid"])
        self.assertIn("cwd", snapshot)

    def test_process_report_contains_identity(self):
        snapshot = {
            "name":"python",
            "pid":1,
            "ppid":0,
            "state":"S",
            "uid":[1000,1000,1000,1000],
            "gid":[1000,1000,1000,1000],
            "groups":[1000],
            "umask":"0022",
            "cwd":"<project-root>",
            "exe":"/usr/bin/python3",
            "cmdline":"python x.py",
            "open_file_descriptors":[],
            "privacy_note":"test",
        }
        report = lab.process_report_markdown(snapshot)
        self.assertIn("UID-uri", report)
        self.assertIn("PID", report)


class DiagramTests(unittest.TestCase):
    def test_dot_contains_authority_chain(self):
        dot = lab.authority_flow_dot()
        for token in (
            "agent",
            "process",
            "identity",
            "kernel",
            "object",
            "allow",
            "deny",
        ):
            self.assertIn(token, dot)

    def test_dot_marks_kernel_boundary_red(self):
        self.assertIn('color="red"', lab.authority_flow_dot())


if __name__ == "__main__":
    unittest.main()
PY

Compilează și rulează:

python -m py_compile   src/lesson05_permissions_lab.py   tests/test_lesson05_permissions.py

python -m unittest -v tests/test_lesson05_permissions.py

Rezultatul verificat pentru această versiune:

Ran 26 tests

OK

Cele 26 de teste acoperă selecția proprietar/grup/ceilalți, capcana în care proprietarul nu primește automat drepturile mai permisive ale grupului, parsarea /proc/<pid>/status, înlocuirea rădăcinii proiectului în căile raportate, validarea politicii, blocarea căilor care ies din sandbox și structura diagramei.

20. Generează diagrama autorității

Programul generează DOT-ul:

python src/lesson05_permissions_lab.py dot \
  --output diagrams/lesson05_authority_flow.dot

dot -Tpng \
  diagrams/lesson05_authority_flow.dot \
  -o diagrams/lesson05_authority_flow.png

file diagrams/lesson05_authority_flow.png
xdg-open diagrams/lesson05_authority_flow.png

Sursa DOT verificată este:

digraph authority_flow {
  rankdir=LR;
  graph [bgcolor="transparent", pad="0.2", nodesep="0.35", ranksep="0.55"];
  node [fontname="DejaVu Sans", fontsize=10, shape=box, style="rounded"];
  edge [fontname="DejaVu Sans", fontsize=9];

  agent [label="Agent / controller\nalege acțiunea"];
  tool [label="Instrument\ntransformă decizia în operație"];
  process [label="Proces Linux\nPID + stare"];
  identity [label="Identitate proces\nEUID / EGID / grupuri"];
  kernel [label="Kernel\nverifică accesul", color="red", penwidth=2];
  object [label="Obiect\nfișier / folder / proces"];
  allow [label="PERMIS", color="green", fontcolor="green"];
  deny [label="REFUZAT", color="red", fontcolor="red"];

  agent -> tool;
  tool -> process;
  process -> identity;
  identity -> kernel;
  object -> kernel [label="proprietar/grup/rwx\n+ reguli suplimentare"];
  kernel -> allow;
  kernel -> deny;
}

Conturul roșu al nodului kernel reprezintă limita la care operația este autorizată sau refuzată. Diagrama nu spune că kernelul este singurul mecanism de securitate; spune că modelul și promptul nu scriu direct pe disc.

21. Completează fișa de autoritate a agentului tău cumulativ

Acum mutăm lecția de la structura artificială de test la proiectul pe care îl construiești în curs.

Creează docs/lessons/lesson-05/authority-design.md:

cat > docs/lessons/lesson-05/authority-design.md <<'EOF'
# Fișa de autoritate — Lecția 5

Completează această fișă pentru agentul tău cumulativ.

## Proces

- Program / entrypoint:
- Cum este pornit:
- Cont de serviciu propus:
- Grupuri necesare:

## Citire

- Foldere / resurse care trebuie citite:
- Foldere / resurse care NU trebuie citite:

## Scriere

- Foldere în care trebuie să scrie:
- Surse care trebuie să rămână doar pentru citire:

## Procese și servicii

- Subprocese pe care le poate porni:
- Procese/servicii pe care le poate opri sau reporni:
- Procese pe care NU trebuie să le controleze:

## Privilegii

- Are nevoie de root? De ce?
- Are nevoie de o capability? Care și de ce?
- Ar putea funcționa cu o regulă mai îngustă?

## Dovada de succes

- Operație care trebuie să reușească:
- Operație care trebuie să fie refuzată:
- Ce vei inspecta pentru a demonstra ambele rezultate:
EOF

Nu accepta răspunsuri precum „acces la tot proiectul” fără justificare. Pentru fiecare autoritate largă, întreabă dacă poate fi redusă la un folder, un grup, o operație sau un subproces delimitat.

22. Înregistrează decizia — ADR-0005

ADR-0001 a stabilit punctul de referință determinist; ADR-0002 bucla cu stare explicită; ADR-0003 responsabilitățile înaintea produselor; ADR-0004 comparația după responsabilitate. ADR-0005 leagă toate acestea de autoritatea reală a procesului.

Creează docs/architecture-decisions/ADR-0005-least-privilege-process-identity.md:

cat > docs/architecture-decisions/ADR-0005-least-privilege-process-identity.md <<'EOF'
# ADR-0005 — Procesele agentice rulează cu privilegiul minim necesar

## Stare

Acceptat pentru proiectul de învățare.

## Context

Lecțiile 2–4 au separat modelul, controllerul, instrumentele și mecanismele de izolare. O acțiune locală este însă executată de un proces al sistemului de operare. Dacă procesul primește mai multe drepturi decât cere sarcina, o eroare de logică, un instrument defect sau o intrare rău intenționată poate afecta resurse fără legătură cu obiectivul agentului.

## Decizie

Procesele agentice nu vor rula implicit ca root. Pentru fiecare proces descriem explicit:

- identitatea de execuție;
- folderele și resursele pe care trebuie să le citească;
- zonele în care poate scrie;
- procesele sau serviciile pe care le poate controla;
- orice privilegiu suplimentar și motivul lui.

Separăm sursele doar pentru citire de zonele de rezultate și preferăm permisiuni, grupuri, ACL-uri sau mecanisme de izolare limitate în locul privilegiului administrativ global.

## Consecințe

- Autoritatea devine auditată și explicabilă.
- Unele operații administrative vor necesita o cale separată și explicită.
- Testele trebuie să includă atât operații permise, cât și operații care trebuie refuzate.
- Configurația de permisiuni devine parte din arhitectură, nu un detaliu reparat cu `sudo`.
- Laboratoarele viitoare pot adăuga containere și sandbox-uri fără să piardă modelul de identitate al procesului.
EOF

23. Verifică proiectul înainte de commit

tree -a -L 4   configs   diagrams   docs/architecture-decisions   docs/lessons/lesson-05   src   tests

git status --short

Confirmă că mediul de test și capturile locale nu intră în Git:

git check-ignore -v   data/lesson05/permission_sandbox   data/lesson05/process_snapshot.json   data/lesson05/permission_probe.json   docs/lessons/lesson-05/process_report.md   docs/lessons/lesson-05/permission_probe_report.md   logs/lesson05/worker.pid

În schimb, politica JSON, codul, testele, fișierele DOT/PNG, ADR-ul și fișa de autoritate trebuie să rămână versionabile.

24. Salvează numai artefactele revizuite

git add \
  .gitignore \
  configs/lesson05_authority_policy.json \
  src/lesson05_permissions_lab.py \
  tests/test_lesson05_permissions.py \
  diagrams/lesson05_authority_flow.dot \
  diagrams/lesson05_authority_flow.png \
  docs/architecture-decisions/ADR-0005-least-privilege-process-identity.md \
  docs/lessons/lesson-05/authority-design.md \
  logs/lesson05/.gitkeep

git diff --cached --stat
git diff --cached

git commit -m "lesson 5: inspect process identity and permissions"

git diff --cached este lung și se poate deschide într-un cititor paginat. Derulezi cu săgețile sau Page Up/Page Down și ieși cu tasta q.

Nu folosi git add . înainte să verifici git status. În această lecție avem în mod intenționat rezultate locale pe care nu vrem să le publicăm.

25. Ce ai construit de fapt

26. Depanare: pornește de la simptom

python3 src/lesson05_permissions_lab.py setup --sandbox /tmp/altloc
EROARE: Refuz să modific această cale. Sandbox-ul trebuie să fie exact data/lesson05/permission_sandbox din proiectul curent.

python3 src/lesson05_permissions_lab.py explain data/lesson05/nuexistă.md read
EROARE: Calea nu există: data/lesson05/nuexistă.md

python3 src/lesson05_permissions_lab.py probe | grep '^| `delete'
| `delete_readonly_in_writable_dir` | permis | permis | ✓ |

git -c user.email="" -c user.name="" commit -m x
Author identity unknown

*** Please tell me who you are.

"warning": "Acesta este numai modelul clasic rwx. Explicația folosește EUID/EGID ca aproximație didactică pentru FSUID/FSGID. Verdictul kernelului poate fi influențat de ACL, capabilities, LSM, opțiuni de montare și alte reguli."

27. Exerciții practice

Exerciții de bază

Exercițiul A — Explică identitatea procesului de test

Folosește ps, /proc/<pid>/status și raportul generat. Scrie trei propoziții: cine este procesul, ce proces l-a pornit și ce credențiale numerice are.

Exercițiul B — Proprietarul nu primește și drepturile grupului

Explică testul test_owner_does_not_accumulate_group_write. De ce ar fi periculos un evaluator care adună drepturile proprietarului cu ale grupului?

Exercițiul C — Demonstrează ștergerea prin folderul părinte

Rulează stat pe fișier și pe folderul părinte, apoi explain-delete și probe. Explică de ce modul 0444 al fișierului nu blochează ștergerea numelui.

Exercițiul D — Schimbă numai folderul de ieșire

După setup, schimbă temporar modul folderului drafts la 0500 și rulează proba. Care singur rezultat ar trebui să se schimbe? Rulează din nou setup pentru reset.

chmod 0500 data/lesson05/permission_sandbox/drafts
ls -ld data/lesson05/permission_sandbox/drafts
dr-x------ 2 ai ai 60 oct 17 11:37 drafts

Exercițiul E — Completează fișa de autoritate

Nu lăsa niciun „acces la tot” fără justificare. Identifică cel puțin o operație care trebuie să reușească și una care trebuie refuzată.

Extensii

Exercițiul F — Adaugă un scenariu de execute

Adaugă în structura de test un script fără bit x și o probă care demonstrează diferența dintre citirea scriptului și executarea lui. Nu folosi sudo.

Exercițiul G — Adaugă un ACL controlat

Într-o structură de test separată, folosește setfacl pentru a adăuga o regulă și demonstrează de ce evaluatorul clasic poate să nu fie suficient. Nu modifica fișiere din afara data/lesson05.

Exercițiul H — Raport de diferență între două procese

Extinde programul cu o comandă care compară două capturi de stare și evidențiază diferențe de EUID/EGID, grupuri, CWD și FD-uri. Capturile de stare trebuie să rămână ignorate de Git.

Exercițiul I — Proiectează un proces auxiliar privilegiat îngust

Fără să îl implementezi, descrie un caz în care agentul are nevoie de o singură operație administrativă. Separă procesul agentului de procesul auxiliar și definește exact datele de intrare acceptate, aprobarea și auditul.

28. Quiz practic

Întrebarea 1

De ce inspectăm un proces pornit de noi și nu un daemon ales la întâmplare?

Întrebarea 2

De ce este ignorat de Git fișierul cu captura de stare a procesului?

Întrebarea 3

Ce dovedește process_snapshot.json și ce nu dovedește?

Întrebarea 4

De ce refuză proba UID 0?

Întrebarea 5

De ce basic_mode_allows() nu trebuie prezentată ca verdictul kernelului?

"warning": "Acesta este numai modelul clasic rwx. Explicația folosește EUID/EGID ca aproximație didactică pentru FSUID/FSGID. Verdictul kernelului poate fi influențat de ACL, capabilities, LSM, opțiuni de montare și alte reguli."

Întrebarea 6

Ce arată FD 3 și FD 4 în procesul de test?

- FD `0` → `/dev/null`
- FD `1` → `<project-root>/logs/lesson05/worker.out`
- FD `2` → `<project-root>/logs/lesson05/worker.out`
- FD `3` → `<project-root>/data/lesson05/permission_sandbox/source/contract-a.txt`
- FD `4` → `<project-root>/data/lesson05/permission_sandbox/drafts/worker-heartbeat.txt`

Întrebarea 7

De ce ștergerea fișierului 0444 poate reuși?

python3 src/lesson05_permissions_lab.py explain-delete data/lesson05/permission_sandbox/source/contract-a.txt
{
  "path": "data/lesson05/permission_sandbox/source/contract-a.txt",
  "parent": "data/lesson05/permission_sandbox/source",
  "parent_write": false,
  "parent_traverse": true,
  "basic_mode_allows_delete": false,

Întrebarea 8

Ce rol are namei -l în acest laborator?

namei -l data/lesson05/permission_sandbox/source/contract-a.txt
f: data/lesson05/permission_sandbox/source/contract-a.txt
drwxr-xr-x ai ai data
drwxr-xr-x ai ai lesson05
drwxr-xr-x ai ai permission_sandbox
dr-xr-xr-x ai ai source
-r--r--r-- ai ai contract-a.txt

Întrebarea 9

Un subproces nou înseamnă automat o identitate mai restrictivă?

sh -c 'id -u'
1000
id -u
1000

Întrebarea 10

Ce legătură are ADR-0005 cu Lecția 4?

29. Checklist de final

• [ ] lucrezi în proiectul cumulativ și mediul .venv este activ;
• [ ] politica JSON are 6 operații așteptate;
• [ ] setup creează numai sandbox-ul permis;
• [ ] poți explica modurile 0555, 0444, 0700 și 0000;
• [ ] ai folosit namei -l pentru calea contractului;
• [ ] procesul de test a avut un PID real și descriptori către source și drafts;
• [ ] ai inspectat UID/GID/Groups/Umask din /proc/<pid>/status;
• [ ] captura de stare locală este ignorată de Git;
• [ ] toate cele șase probe corespund politicii;
• [ ] poți explica de ce fișierul doar pentru citire poate fi șters;
• [ ] programul refuză probe ca root;
• [ ] toate cele 26 de teste trec;
• [ ] DOT-ul și PNG-ul sunt generate;
• [ ] fișa de autoritate este completată;
• [ ] ADR-0005 există;
• [ ] ai inspectat diferențele pregătite pentru commit cu git diff --cached.

30. Verificarea executată pentru această versiune

Acest bloc documentează verificările refăcute după auditul tehnic și lingvistic al acestei versiuni.

• Python: programul și testele trec verificarea de sintaxă cu py_compile.
• Teste automate: 26 din 26 trec.
• Probă reală neprivilegiată: toate cele șase operații corespund politicii declarate.
• Graphviz: fișierul DOT produce un PNG valid.
• Validarea politicii: versiunea de schemă necunoscută, o intrare obligatorie lipsă și căile care conțin .. sunt refuzate; fișierele, JSON-ul sau căile lipsă produc mesaje scurte, fără traceback. Protecția sandbox-ului respinge și o cale arbitrară sau una „asemănătoare” din altă zonă.
• Protecția root: comenzile didactice care ar putea modifica mediul sau produce explicații înșelătoare sub UID 0 refuză rularea ca root.
• Procesul de test își publică PID-ul numai după ce a deschis fișierele pe care vrem să le observăm, iar pasul de pornire șterge mai întâi orice PID vechi rămas dintr-o rulare anterioară.

31. Ce păstrăm pentru Lecția 6

Laboratorul 5 a folosit câteva comenzi Linux ca instrumente de măsurare, dar nu a încercat să predea terminalul complet. Ai acum motive concrete pentru a învăța următoarele lucruri: cum sunt interpretate căile, cum legi comenzi prin pipe-uri, cum citești variabile de mediu, cum inspectezi servicii și loguri și cum lucrezi la distanță prin SSH.

Artefactele care merg mai departe sunt:

configs/lesson05_authority_policy.json
src/lesson05_permissions_lab.py
tests/test_lesson05_permissions.py
docs/lessons/lesson-05/authority-design.md
docs/architecture-decisions/ADR-0005-least-privilege-process-identity.md
diagrams/lesson05_authority_flow.dot
diagrams/lesson05_authority_flow.png