Private JSON

1"""Bounded private JSON evidence reader shared by retained-operation workflows."""
2from __future__ import annotations
3
4import json
5import os
6import stat
7from pathlib import Path
8
9try:
10    from . import _bootstrap_io as private_io
11except ImportError:
12    import _bootstrap_io as private_io
13
14
15def _json(raw):
16    def unique(pairs):
17        value = {}
18        for key, child in pairs:
19            if key in value:
20                raise ValueError("Duplicate field")
21            value[key] = child
22        return value
23    try:
24        value = json.loads(raw, object_pairs_hook=unique)
25        json.dumps(value, allow_nan=False, ensure_ascii=False).encode("utf-8")
26        if not isinstance(value, dict):
27            raise ValueError("Expected object")
28        return value
29    except (ValueError, UnicodeError, RecursionError) as exc:
30        raise private_io.failure("recheck-evidence-invalid", "Retain bounded, unmodified UTF-8 object evidence.") from exc
31
32
33def read_private(path):
34    path = Path(path)
35    private_io.private_directory(str(path.parent))
36    try:
37        selected = path.lstat()
38        if (not stat.S_ISREG(selected.st_mode) or selected.st_nlink != 1
39                or getattr(selected, "st_file_attributes", 0) & 0x400):
40            raise OSError("Not an ordinary private file")
41        if os.name == "nt":
42            private_io._windows_private(path)
43        elif selected.st_uid != os.getuid() or stat.S_IMODE(selected.st_mode) & 0o077:
44            raise OSError("Not private")
45        descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
46        try:
47            handle = os.fdopen(descriptor, "rb")
48        except BaseException as error:
49            try:
50                os.close(descriptor)
51            except OSError:
52                error.add_note("Private evidence descriptor cleanup could not be confirmed.")
53            raise
54        with handle:
55            opened = os.fstat(handle.fileno())
56            if (selected.st_dev, selected.st_ino) != (opened.st_dev, opened.st_ino):
57                raise OSError("Evidence changed identity")
58            raw = handle.read(private_io.MAX_BYTES + 1)
59        if len(raw) > private_io.MAX_BYTES:
60            raise OSError("Evidence exceeds bound")
61        return _json(raw.decode("utf-8"))
62    except (OSError, UnicodeError) as exc:
63        raise private_io.failure("recheck-evidence-unreadable", "Select existing private, unlinked evidence files; no permissions were changed.") from exc