Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions debian/changelog
Original file line number Diff line number Diff line change
@@ -1,3 +1,19 @@
linuxmuster-linuxclient7 (7.4.0) lmn74; urgency=high

* Security: validate the path and share name passed to the sudoTools mount-share
helper to prevent path traversal / arbitrary CIFS mounts as root
* Security: restrict install-printer to safe device-uri schemes and validate the
printer name (sudoTools is callable NOPASSWD by any domain user)
* Security: escape all values interpolated into LDAP search filters to prevent
LDAP filter injection
* Feat: create user homedirs with umask=077 so users cannot read each other's data (#44)
* Fix: printers are not uninstalled on logout because lpadmin is not in PATH (#93)
* Fix: onBoot hook could block the boot process when the DC/sysvol is unreachable (#61)
* Compat: support Ubuntu 26.04 / Samba 4.2x (drop deprecated "client use spnego",
robust SSSD version parsing, timeout on mount.cifs)

-- Lukas Lukic-Spitznagel <lukas.spitznagel@netzint.de> Thu, 23 Jul 2026 12:00:00 +0200

linuxmuster-linuxclient7 (1.1.4) lmn73; urgency=medium

* Fix: regressions in setup and prepareImage (#91)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ def readAttributes():
:return: Tuple (success, dict of attributes)
:rtype: tuple
"""
return ldapHelper.searchOne(f"(sAMAccountName={hostname()}$)")
return ldapHelper.searchOne(f"(sAMAccountName={ldapHelper.escape(hostname())}$)")

def isInGroup(groupName):
"""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,9 @@

shareMountBasepath = "/home/{}/media"
hiddenShareMountBasepath = "/srv/samba/{}"
# Timeout (in seconds) for the mount.cifs call, so an unreachable server
# cannot block boot / login indefinitely (see issue #61).
mountTimeoutSeconds = 10
machineAccountSysvolMountPath = "/var/lib/samba/sysvol"
defaultShareLetterTemplate = " ({letter}:)"

Expand Down
4 changes: 2 additions & 2 deletions usr/lib/python3/dist-packages/linuxmusterLinuxclient7/gpo.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ def _findApplicablePolicies():
policyName = f"sophomorix:school:{schoolName}"

# find policy
rc, policyAdObject = ldapHelper.searchOne(f"(displayName={policyName})")
rc, policyAdObject = ldapHelper.searchOne(f"(displayName={ldapHelper.escape(policyName)})")
if not rc:
return False, None

Expand All @@ -99,7 +99,7 @@ def _parsePolicy(policyDn):
"""

# Find policy in AD
rc, policyAdObject = ldapHelper.searchOne(f"(distinguishedName={policyDn[0]})")
rc, policyAdObject = ldapHelper.searchOne(f"(distinguishedName={ldapHelper.escape(policyDn[0])})")
if not rc:
logging.error("===> Could not find poilcy in AD! ===")
return False
Expand Down
Original file line number Diff line number Diff line change
@@ -1,8 +1,20 @@
import ldap, ldap.sasl, sys, getpass, subprocess
import ldap, ldap.sasl, ldap.filter, sys, getpass, subprocess
from linuxmusterLinuxclient7 import logging, constants, config, user, computer

_currentLdapConnection = None

def escape(value):
"""
Escapes a value so it can be safely embedded into an LDAP search filter.
This prevents LDAP filter injection (e.g. via crafted usernames or hostnames).

:param value: The value to escape
:type value: str
:return: The escaped value
:rtype: str
"""
return ldap.filter.escape_filter_chars(str(value))

def serverUrl():
"""
Returns the server URL
Expand Down Expand Up @@ -109,7 +121,7 @@ def isObjectInGroup(objectDn, groupName):
:rtype: bool
"""
logging.debug(f"= Testing if object {objectDn} is a member of group {groupName} =")
rc, groupAdObject = searchOne(f"(&(member:1.2.840.113556.1.4.1941:={objectDn})(sAMAccountName={groupName}))")
rc, groupAdObject = searchOne(f"(&(member:1.2.840.113556.1.4.1941:={escape(objectDn)})(sAMAccountName={escape(groupName)}))")
logging.debug(f"=> Result: {rc} =")
return rc

Expand Down
40 changes: 36 additions & 4 deletions usr/lib/python3/dist-packages/linuxmusterLinuxclient7/printers.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,17 @@
import os, subprocess, re
import os, subprocess, re, shutil
from linuxmusterLinuxclient7 import logging, user

# Resolve absolute paths at import time. When called from the logout PAM hook the
# PATH is reduced to /bin:/usr/bin, so lpadmin (/usr/sbin) is not found otherwise.
# See issue #93.
_lpadminPath = shutil.which("lpadmin") or "/usr/sbin/lpadmin"
_lpstatPath = shutil.which("lpstat") or "/usr/bin/lpstat"

# CUPS device-uri schemes we allow to be installed via the (root) sudoTools helper.
# Notably excludes "file://", which would let an unprivileged domain user write to
# an arbitrary file as root. See security audit.
_allowedPrinterUriSchemes = ("ipp", "ipps", "http", "https", "smb", "lpd", "socket", "dnssd")

def installPrinter(networkPath, name=None, username=None):
"""
Installs a networked printer for a user
Expand Down Expand Up @@ -70,8 +81,16 @@ def translateSambaToIpp(networkPath):
# --------------------

def _installPrinter(username, networkPath, name):
if not _printerNetworkPathIsValid(networkPath):
logging.error(f"Refusing to install printer: invalid network path '{networkPath}'")
return False

if not _printerNameIsValid(name):
logging.error(f"Refusing to install printer: invalid printer name '{name}'")
return False

logging.info(f"Install Printer {name} on {networkPath}")
installCommand = ["timeout", "10", "lpadmin", "-p", name, "-E", "-v", networkPath, "-m", "everywhere", "-u", f"allow:{username}"]
installCommand = ["timeout", "10", _lpadminPath, "-p", name, "-E", "-v", networkPath, "-m", "everywhere", "-u", f"allow:{username}"]
logging.debug(f"* running '{' '.join(installCommand)}'")
p = subprocess.call(installCommand, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if p == 0:
Expand All @@ -88,7 +107,7 @@ def _installPrinterWithoutRoot(networkPath, name):

def _getInstalledPrintersOfUser(username):
logging.info(f"Getting installed printers of {username}")
command = ["lpstat", "-U", username, "-p"]
command = [_lpstatPath, "-U", username, "-p"]
#logging.debug(f"running '{command}'")

result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
Expand All @@ -113,7 +132,7 @@ def _getInstalledPrintersOfUser(username):

def _uninstallPrinter(name):
logging.info(f"Uninstall Printer {name}")
uninstallCommand = ["timeout", "10", "lpadmin", "-x", name]
uninstallCommand = ["timeout", "10", _lpadminPath, "-x", name]
logging.debug(f"* running '{' '.join(uninstallCommand)}'")
p = subprocess.call(uninstallCommand, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if p == 0:
Expand All @@ -124,3 +143,16 @@ def _uninstallPrinter(name):
else:
logging.fatal(f"* Error Uninstalling Printer {name}!")
return False

def _printerNetworkPathIsValid(networkPath):
if not networkPath or "\x00" in networkPath or "\n" in networkPath:
return False
match = re.match(r"^([a-zA-Z][a-zA-Z0-9+.-]*)://", networkPath)
if not match:
return False
return match.group(1).lower() in _allowedPrinterUriSchemes

def _printerNameIsValid(name):
if not name:
return False
return re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", name) is not None
10 changes: 8 additions & 2 deletions usr/lib/python3/dist-packages/linuxmusterLinuxclient7/realm.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,14 +69,20 @@ def isJoined():
else:
return len(joinedDomains) > 0

def pullKerberosTicketForComputerAccount():
def pullKerberosTicketForComputerAccount(timeoutSeconds=None):
"""
Pulls a kerberos ticket using the computer account from `/etc/krb5.keytab`

:param timeoutSeconds: If set, abort the kinit after this many seconds so an
unreachable domain controller cannot block the caller (e.g. boot). See issue #61.
:type timeoutSeconds: int, optional
:return: True on success, False otherwise
:rtype: bool
"""
return subprocess.call(["kinit", "-k", computer.krbHostName()]) == 0
command = ["kinit", "-k", computer.krbHostName()]
if timeoutSeconds is not None:
command = ["timeout", str(timeoutSeconds)] + command
return subprocess.call(command) == 0

def verifyDomainJoin():
"""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -374,7 +374,12 @@ def _deleteObsoleteFiles():
return True

def _parseVersion(versionString):
return tuple(int(p) for p in versionString.split("."))
# Only look at the leading dotted-number part so packaging suffixes emitted by
# newer distributions (e.g. "2.10.0-1ubuntu1" on Ubuntu 26.04) do not break parsing.
match = re.match(r"\d+(?:\.\d+)*", versionString.strip())
if not match:
raise ValueError(f"Could not parse version string: {versionString!r}")
return tuple(int(p) for p in match.group(0).split("."))

def _getSssdVersion():
try:
Expand Down
58 changes: 57 additions & 1 deletion usr/lib/python3/dist-packages/linuxmusterLinuxclient7/shares.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,8 +138,28 @@ def getLocalSysvolPath():
# This parameter influences the `cruid` mount option.
def _mountShare(username, networkPath, shareName, hiddenShare, useCruidOfExecutingUser=False):

# Normalize and validate the untrusted inputs before doing anything as root.
# _mountShare() is reachable by any domain user via the sudoTools helper, so
# unvalidated values would allow mounting an arbitrary CIFS server to an
# arbitrary location (path traversal via shareName). See security audit.
networkPath = networkPath.replace("\\", "/")

if not _networkPathIsValid(networkPath):
logging.error(f"Refusing to mount: invalid network path '{networkPath}'")
return False, None

if not _shareNameIsValid(shareName):
logging.error(f"Refusing to mount: invalid share name '{shareName}'")
return False, None

mountpoint = _getShareMountpoint(networkPath, username, hiddenShare, shareName)

# Defense in depth: make sure the resolved mountpoint really stays inside the
# per-user base directory and did not escape it through traversal.
if not _mountpointIsWithinBase(mountpoint, username, hiddenShare):
logging.error(f"Refusing to mount: mountpoint '{mountpoint}' escapes the allowed base directory")
return False, None

mountCommandOptions = f"file_mode=0700,dir_mode=0700,sec=krb5,nodev,nosuid,mfsymlinks,nobrl,vers=3.0,user={username}"
rc, networkConfig = config.network()
domain = None
Expand All @@ -162,7 +182,10 @@ def _mountShare(username, networkPath, shareName, hiddenShare, useCruidOfExecuti
gid = -1
logging.warning("Uid could not be found! Continuing anyway!")

mountCommand = [shutil.which("mount.cifs"), "-o", mountCommandOptions, networkPath, mountpoint]
mountBinary = shutil.which("mount.cifs") or "/usr/sbin/mount.cifs"
# Wrap the mount in `timeout` so an unreachable server cannot block
# boot / login indefinitely (issue #61).
mountCommand = ["timeout", str(constants.mountTimeoutSeconds), mountBinary, "-o", mountCommandOptions, networkPath, mountpoint]

logging.debug(f"Trying to mount '{networkPath}' to '{mountpoint}'")
logging.debug("* Creating directory...")
Expand Down Expand Up @@ -259,3 +282,36 @@ def _getShareMountpoint(networkPath, username, hidden, shareName = None):

def _directoryIsMountpoint(dir):
return subprocess.call(["mountpoint", "-q", dir]) == 0

def _networkPathIsValid(networkPath):
"""
Validates a CIFS network path of the form `//server/share[/subdir...]`.
Rejects empty paths, path traversal and control characters.
"""
if not networkPath or "\x00" in networkPath or "\n" in networkPath:
return False
if ".." in networkPath.split("/"):
return False
return re.fullmatch(r"//[^/\x00]+/[^\x00]+", networkPath) is not None

def _shareNameIsValid(shareName):
"""
Validates the share name, which becomes a single directory below the user's
mount base. It must therefore be a single path component without any slashes
or traversal.
"""
if not shareName or shareName in (".", ".."):
return False
if "/" in shareName or "\\" in shareName or "\x00" in shareName:
return False
return True

def _mountpointIsWithinBase(mountpoint, username, hidden):
if hidden:
base = constants.hiddenShareMountBasepath.format(username)
else:
base = constants.shareMountBasepath.format(username)

base = os.path.normpath(base)
normalized = os.path.normpath(mountpoint)
return normalized == base or normalized.startswith(base + "/")
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
from unittest import mock
from .. import ldapHelper


def test_escape():
# Plain values are unchanged
assert ldapHelper.escape("user1") == "user1"
# LDAP filter metacharacters are escaped so they cannot alter the filter
assert ldapHelper.escape("*") == r"\2a"
assert ldapHelper.escape("a)(b") == r"a\29\28b"
assert ldapHelper.escape("a\\b") == r"a\5cb"
# Non-str input is handled
assert ldapHelper.escape(123) == "123"


@mock.patch("linuxmusterLinuxclient7.ldapHelper.searchOne")
def test_isObjectInGroupEscapesFilter(mockSearchOne):
mockSearchOne.return_value = (True, {})
# A crafted group name with filter metacharacters must be escaped, not injected.
assert ldapHelper.isObjectInGroup("cn=user,dc=x", "domain users)(cn=*")
usedFilter = mockSearchOne.call_args.args[0]
assert "*" not in usedFilter
assert r"\2a" in usedFilter
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,41 @@
from unittest import mock
from .. import printers

def test_printerNetworkPathIsValid():
assert printers._printerNetworkPathIsValid("ipp://server/printers/P1")
assert printers._printerNetworkPathIsValid("smb://server/P1")
assert printers._printerNetworkPathIsValid("https://server/printers/P1")
# dangerous / invalid schemes must be rejected
assert not printers._printerNetworkPathIsValid("file:///etc/passwd")
assert not printers._printerNetworkPathIsValid("/etc/passwd")
assert not printers._printerNetworkPathIsValid("")
assert not printers._printerNetworkPathIsValid("ipp://server/\nP1")


def test_printerNameIsValid():
assert printers._printerNameIsValid("EW-farblaser")
assert printers._printerNameIsValid("printer1")
assert not printers._printerNameIsValid("")
assert not printers._printerNameIsValid("-flag")
assert not printers._printerNameIsValid("a/b")
assert not printers._printerNameIsValid("a b")


@mock.patch("subprocess.call")
@mock.patch("linuxmusterLinuxclient7.hooks.user.isRoot")
@mock.patch("linuxmusterLinuxclient7.hooks.user.username")
def test_installPrinterRejectsUnsafeInput(mockUserUsername, mockUserIsRoot, mockSubprocessCall):
mockUserIsRoot.return_value = True
mockUserUsername.return_value = "user1"

# Unsafe device-uri scheme (file://) must be rejected without running lpadmin.
assert not printers.installPrinter("file:///etc/passwd", "printer1")
# Unsafe printer name must be rejected too.
assert not printers.installPrinter("ipp://server/printers/P1", "../evil")
# The privileged lpadmin command (wrapped in `timeout`) must never have run.
assert _getCallsTo(mockSubprocessCall, "timeout") == []


def test_translateSambaToIpp():
assert printers.translateSambaToIpp("\\\\linuxmuster.lan\\printer1") == (True, "ipp://linuxmuster.lan/printers/printer1")
assert printers.translateSambaToIpp("\\\\linuxmuster.lan\\") == (False, None)
Expand All @@ -18,13 +53,13 @@ def test_installPrinter(mockUserUsername, mockUserIsRoot, mockSubprocessCall):
mockUserUsername.return_value = "user1"

assert printers.installPrinter("ipp://linuxmuster.lan/printers/printer1")
assert _getCallsTo(mockSubprocessCall, "timeout")[-1] == ["timeout", "10", "lpadmin", "-p", "printer1", "-E", "-v", "ipp://linuxmuster.lan/printers/printer1", "-m", "everywhere", "-u", "allow:user1"]
assert _getCallsTo(mockSubprocessCall, "timeout")[-1] == ["timeout", "10", printers._lpadminPath, "-p", "printer1", "-E", "-v", "ipp://linuxmuster.lan/printers/printer1", "-m", "everywhere", "-u", "allow:user1"]

assert printers.installPrinter("ipp://linuxmuster.lan/printers/printer1", "printer2")
assert _getCallsTo(mockSubprocessCall, "timeout")[-1] == ["timeout", "10", "lpadmin", "-p", "printer2", "-E", "-v", "ipp://linuxmuster.lan/printers/printer1", "-m", "everywhere", "-u", "allow:user1"]
assert _getCallsTo(mockSubprocessCall, "timeout")[-1] == ["timeout", "10", printers._lpadminPath, "-p", "printer2", "-E", "-v", "ipp://linuxmuster.lan/printers/printer1", "-m", "everywhere", "-u", "allow:user1"]

assert printers.installPrinter("ipp://linuxmuster.lan/printers/printer1", "printer2", "user2")
assert _getCallsTo(mockSubprocessCall, "timeout")[-1] == ["timeout", "10", "lpadmin", "-p", "printer2", "-E", "-v", "ipp://linuxmuster.lan/printers/printer1", "-m", "everywhere", "-u", "allow:user2"]
assert _getCallsTo(mockSubprocessCall, "timeout")[-1] == ["timeout", "10", printers._lpadminPath, "-p", "printer2", "-E", "-v", "ipp://linuxmuster.lan/printers/printer1", "-m", "everywhere", "-u", "allow:user2"]

mockUserIsRoot.return_value = False
assert printers.installPrinter("ipp://linuxmuster.lan/printers/printer1")
Expand Down Expand Up @@ -57,8 +92,8 @@ def test_uninstallAllPrintersOfUser(mockSubprocessRun, mockSubprocessCall):
mockSubprocessRun.return_value = CompletedProcess(args=["lpstat", "-U", "user1", "-p"], returncode=0, stdout=lpstatStdout)

assert printers.uninstallAllPrintersOfUser("user1")
assert _getCallsTo(mockSubprocessRun, "lpstat")[-1] == ['lpstat', '-U', 'user1', '-p']
assert _getCallsTo(mockSubprocessCall, "timeout") == [["timeout", "10", "lpadmin", "-x", "printer1"], ["timeout", "10", "lpadmin", "-x", "printer2"]]
assert _getCallsTo(mockSubprocessRun, printers._lpstatPath)[-1] == [printers._lpstatPath, '-U', 'user1', '-p']
assert _getCallsTo(mockSubprocessCall, "timeout") == [["timeout", "10", printers._lpadminPath, "-x", "printer1"], ["timeout", "10", printers._lpadminPath, "-x", "printer2"]]

mockSubprocessRun.return_value = CompletedProcess(args=["lpstat", "-U", "user1", "-p"], returncode=1)
assert printers.uninstallAllPrintersOfUser("user1")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,14 @@ def test_pullKerberosTicketForComputerAccount(mockSubprocessCall):
mockSubprocessCall.return_value = 1
assert not realm.pullKerberosTicketForComputerAccount()

@mock.patch("subprocess.call")
@mock.patch("linuxmusterLinuxclient7.computer.krbHostName", lambda : "linuxmuster.lan")
def test_pullKerberosTicketForComputerAccountWithTimeout(mockSubprocessCall):
mockSubprocessCall.return_value = 0
assert realm.pullKerberosTicketForComputerAccount(timeoutSeconds=5)
calls = _getCallsTo(mockSubprocessCall, "timeout")
assert ["timeout", "5", "kinit", "-k", "linuxmuster.lan"] in calls

@mock.patch("subprocess.call")
@mock.patch("linuxmusterLinuxclient7.realm.isJoined")
@mock.patch("linuxmusterLinuxclient7.realm.pullKerberosTicketForComputerAccount")
Expand Down
Loading
Loading