mirror of
https://github.com/Kozea/Radicale.git
synced 2025-09-15 20:36:55 +00:00
Improve: user/group retrievement for running service and directories
This commit is contained in:
parent
001d44faae
commit
ca3fd9a3ff
2 changed files with 63 additions and 16 deletions
|
@ -31,7 +31,7 @@ import threading
|
|||
from tempfile import TemporaryDirectory
|
||||
from typing import Iterator, Type, Union
|
||||
|
||||
from radicale import storage, types
|
||||
from radicale import storage, types, utils
|
||||
|
||||
if sys.platform == "win32":
|
||||
import ctypes
|
||||
|
@ -320,13 +320,36 @@ def name_from_path(path: str, collection: "storage.BaseCollection") -> str:
|
|||
|
||||
def path_permissions(path):
|
||||
path = pathlib.Path(path)
|
||||
return [path.owner(), path.group(), path.stat().st_mode]
|
||||
|
||||
try:
|
||||
uid = utils.unknown_if_empty(path.stat().st_uid)
|
||||
except (KeyError, NotImplementedError):
|
||||
uid = "UNKNOWN"
|
||||
|
||||
try:
|
||||
gid = utils.unknown_if_empty(path.stat().st_gid)
|
||||
except (KeyError, NotImplementedError):
|
||||
gid = "UNKNOWN"
|
||||
|
||||
try:
|
||||
mode = utils.unknown_if_empty("%o" % path.stat().st_mode)
|
||||
except (KeyError, NotImplementedError):
|
||||
mode = "UNKNOWN"
|
||||
|
||||
try:
|
||||
owner = utils.unknown_if_empty(path.owner())
|
||||
except (KeyError, NotImplementedError):
|
||||
owner = "UNKNOWN"
|
||||
|
||||
try:
|
||||
group = utils.unknown_if_empty(path.group())
|
||||
except (KeyError, NotImplementedError):
|
||||
group = "UNKNOWN"
|
||||
|
||||
return [owner, uid, group, gid, mode]
|
||||
|
||||
|
||||
def path_permissions_as_string(path):
|
||||
try:
|
||||
pp = path_permissions(path)
|
||||
s = "path=%r owner=%s group=%s mode=%o" % (path, pp[0], pp[1], pp[2])
|
||||
except (KeyError, NotImplementedError):
|
||||
s = "path=%r owner=UNKNOWN(unsupported on this system)" % (path)
|
||||
s = "path=%r owner=%s(%s) group=%s(%s) mode=%s" % (path, pp[0], pp[1], pp[2], pp[3], pp[4])
|
||||
return s
|
||||
|
|
|
@ -226,25 +226,49 @@ def ssl_get_protocols(context):
|
|||
return protocols
|
||||
|
||||
|
||||
def unknown_if_empty(value):
|
||||
if value == "":
|
||||
return "UNKNOWN"
|
||||
else:
|
||||
return value
|
||||
|
||||
|
||||
def user_groups_as_string():
|
||||
if sys.platform != "win32":
|
||||
euid = os.geteuid()
|
||||
egid = os.getegid()
|
||||
try:
|
||||
username = pwd.getpwuid(euid)[0]
|
||||
user = "%s(%d)" % (unknown_if_empty(username), euid)
|
||||
except Exception:
|
||||
# name of user not found
|
||||
s = "user=(%d) group=(%d)" % (euid, egid)
|
||||
return s
|
||||
gids = os.getgrouplist(username, egid)
|
||||
user = "UNKNOWN(%d)" % euid
|
||||
|
||||
egid = os.getegid()
|
||||
groups = []
|
||||
try:
|
||||
gids = os.getgrouplist(username, egid)
|
||||
for gid in gids:
|
||||
try:
|
||||
gi = grp.getgrgid(gid)
|
||||
groups.append("%s(%d)" % (gi.gr_name, gid))
|
||||
groups.append("%s(%d)" % (unknown_if_empty(gi.gr_name), gid))
|
||||
except Exception:
|
||||
groups.append("%s(%d)" % (gid, gid))
|
||||
s = "user=%s(%d) groups=%s" % (username, euid, ','.join(groups))
|
||||
groups.append("UNKNOWN(%d)" % gid)
|
||||
except Exception:
|
||||
try:
|
||||
groups.append("%s(%d)" % (grp.getgrnam(egid)[0], egid))
|
||||
except Exception:
|
||||
# workaround to get groupid by name
|
||||
groups_all = grp.getgrall()
|
||||
found = False
|
||||
for entry in groups_all:
|
||||
if entry[2] == egid:
|
||||
groups.append("%s(%d)" % (unknown_if_empty(entry[0]), egid))
|
||||
found = True
|
||||
break
|
||||
if not found:
|
||||
groups.append("UNKNOWN(%d)" % egid)
|
||||
|
||||
s = "user=%s groups=%s" % (user, ','.join(groups))
|
||||
else:
|
||||
username = os.getlogin()
|
||||
s = "user=%s" % (username)
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue