1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
"""Drop max25d privileges after hardware stack is up (Linux)."""
from __future__ import annotations
import configparser
import grp
import os
import pwd
import sys
from dataclasses import dataclass
from typing import Optional
@dataclass
class RunAsConfig:
user: str = ""
group: str = ""
uid: Optional[int] = None
gid: Optional[int] = None
def _parse_optional_int(raw: str) -> Optional[int]:
text = str(raw).strip()
if not text:
return None
return int(text)
def parse_run_as(cp: configparser.ConfigParser) -> RunAsConfig:
cfg = RunAsConfig()
if not cp.has_section("daemon"):
return cfg
sec = cp["daemon"]
cfg.user = sec.get("user", fallback="").strip()
cfg.group = sec.get("group", fallback="").strip()
if cp.has_option("daemon", "uid"):
cfg.uid = _parse_optional_int(sec.get("uid", fallback=""))
if cp.has_option("daemon", "gid"):
cfg.gid = _parse_optional_int(sec.get("gid", fallback=""))
return cfg
def run_as_configured(cfg: RunAsConfig) -> bool:
return bool(
cfg.user
or cfg.group
or cfg.uid is not None
or cfg.gid is not None
)
def resolve_run_as(cfg: RunAsConfig) -> tuple[int, int, str]:
pw: pwd.struct_passwd | None = None
name = cfg.user
if cfg.user:
pw = pwd.getpwnam(cfg.user)
uid = pw.pw_uid
gid = pw.pw_gid
elif cfg.uid is not None:
pw = pwd.getpwuid(cfg.uid)
uid = pw.pw_uid
gid = pw.pw_gid
name = pw.pw_name
else:
raise ValueError("[daemon] user= or uid= required for privilege drop")
if cfg.uid is not None:
uid = cfg.uid
if cfg.group:
gid = grp.getgrnam(cfg.group).gr_gid
elif cfg.gid is not None:
gid = cfg.gid
return uid, gid, name
def drop_privileges(cfg: RunAsConfig, *, log) -> None:
if os.name != "posix":
return
if not run_as_configured(cfg):
if os.geteuid() == 0:
log(
"running as root after stack start — set [daemon] user= to drop",
area="privilege",
)
return
if os.geteuid() != 0:
return
uid, gid, name = resolve_run_as(cfg)
os.initgroups(name, gid)
os.setgid(gid)
os.setuid(uid)
if os.geteuid() == 0:
raise OSError("privilege drop failed — still root")
log(
f"dropped privileges to uid={uid} gid={gid} ({name})",
area="privilege",
)
def drop_privileges_or_exit(cfg: RunAsConfig, *, log) -> None:
try:
drop_privileges(cfg, log=log)
except (KeyError, ValueError, OSError) as exc:
log(f"privilege drop failed: {exc}", area="privilege")
raise SystemExit(1) from exc
|
