#!/usr/bin/env python3
# Copyright © 2020 Xavier G. <xavier.combinedfs@kindwolf.org>
# This work is free. You can redistribute it and/or modify it under the
# terms of the Do What The Fuck You Want To Public License, Version 2,
# as published by Sam Hocevar. See the COPYING file for more details.
import os
import re
import sys
import stat
import yaml
import errno
import argparse
import threading
# Excerpt from `apt show python3-fusepy`:
# Due to a name clash with the existing API-incompatible python-fuse package,
# the importable module name for fusepy in Debian is 'fusepy' instead of
# upstream's 'fuse'.
try:
from fusepy import FUSE, FuseOSError, Operations
except ImportError:
from fuse import FUSE, FuseOSError, Operations
DEFAULT_ROOT = '/etc/letsencrypt/live'
DEFAULT_CERT_FILTER = False
DEFAULT_WHITELIST = True
DEFAULT_CERT_PATTERN = '.'
DEFAULT_SEPARATOR = '/'
DEFAULT_UID = 0
DEFAULT_GID = 0
DEFAULT_DIR_MODE = 0o555
DEFAULT_REG_MODE = 0o444
DEFAULT_KEY_MODE = 0o400
DEFAULT_SENSITIVE_PATTERN = '/privkey.pem$'
RELOAD_PATH = '/reload'
RELOAD_MSG_OK = b'reload ok\n'
RELOAD_MSG_FAIL = b'reload fail\n'
RELOAD_FILESIZE = max(len(RELOAD_MSG_OK), len(RELOAD_MSG_FAIL))
TIME_PROPS = ('st_atime', 'st_ctime', 'st_mtime')
def read_mode_setting(obj, key, default):
try:
return int(obj[key], 8)
except (KeyError, ValueError):
return default
class CombinedFSConfiguration(object):
def __init__(self, conf_path):
self.read_conf(conf_path)
def read_conf(self, conf_path):
with open(conf_path) as conf_file:
conf = yaml.load(conf_file.read())
self.apply_conf(conf)
self.path = conf_path
def apply_conf(self, conf):
self.root = conf.get('letsencrypt_live', DEFAULT_ROOT)
self.filter = conf.get('cert_filter', DEFAULT_CERT_FILTER)
self.whitelist = conf.get('cert_whitelist', DEFAULT_WHITELIST)
self.pattern = conf.get('cert_pattern', DEFAULT_CERT_PATTERN)
self.separator = conf.get('separator', DEFAULT_SEPARATOR)
self.files = conf.get('files', {})
self.uid = int(conf.get('uid', DEFAULT_UID))
self.gid = int(conf.get('gid', DEFAULT_GID))
self.same_uid_as = conf.get('same-uid-as', None)
self.same_gid_as = conf.get('same-gid-as', None)
self.dir_mode = read_mode_setting(conf, 'dir_mode', DEFAULT_DIR_MODE)
self.reg_mode = read_mode_setting(conf, 'reg_mode', DEFAULT_REG_MODE)
self.key_mode = read_mode_setting(conf, 'key_mode', DEFAULT_KEY_MODE)
self.sensitive_pattern = conf.get('sensitive_pattern', DEFAULT_SENSITIVE_PATTERN)
# Compile regexes:
self.sensitive_pattern_re = re.compile(self.sensitive_pattern)
if self.filter:
self.pattern_re = re.compile(self.pattern)
# Helpers:
def filter_cert(self, cert):
if not self.filter:
return True
return bool(self.pattern_re.match(cert)) == self.whitelist
def analyse_path(self, path):
"""
Return a tuple of three values reflecting what the given path points to.
Raise a FuseOSError with ENOENT if the path does not match anything.
The three values are:
cert: the target certificate; None only for the root directory;
filename: the requested filename; None only for:
- the root directory;
- the cert-specific directory when '/' is used as separator;
file_spec: the specification for the requested filename.
"""
# Initial values:
cert = None
filename = None
file_spec = None
# Root directory:
if path == '/':
return cert, filename, file_spec
# cert and flilename:
path_components = path[1:].split('/')
if self.separator == '/':
if len(path_components) > 2:
raise FuseOSError(errno.ENOENT)
cert = path_components[0]
if len(path_components) == 2:
filename = path_components[1]
else:
if len(path_components) != 1:
raise FuseOSError(errno.ENOENT)
components = path_components[0].split(self.separator)
if len(components) != 2:
raise FuseOSError(errno.ENOENT)
cert = components[0]
filename = components[1]
# Ensure cert is not filtered out:
if not self.filter_cert(cert):
raise FuseOSError(errno.ENOENT)
# file_spec:
if filename is not None:
try:
file_spec = self.files[filename]
except KeyError:
raise FuseOSError(errno.ENOENT)
return cert, filename, file_spec
def expand_path(self, cert, path):
expanded_path = path.replace('${cert}', cert)
if not expanded_path.startswith('/'):
expanded_path = os.path.join(self.root, cert, expanded_path)
return expanded_path
def expand_paths(self, cert, paths):
expanded_paths = []
for path in paths:
method = self.expand_paths if type(path) is list else self.expand_path
expanded_paths.append(method(cert, path))
return expanded_paths
def get_paths(self, cert, file_spec):
return self.expand_paths(cert, file_spec.get('content', []))
def is_sensitive_file(self, filepath):
return self.sensitive_pattern_re.search(filepath)
class CombinedFS(Operations):
def __init__(self, conf_path):
# Configuration:
self.configuration = CombinedFSConfiguration(conf_path)
# File descriptor management:
self.filedesc_lock = threading.Lock()
self.filedesc_index = 0
self.filedesc = {}
# Helpers:
def attributes(self, full_path):
st = os.lstat(full_path)
return dict((key, getattr(st, key)) for key in ('st_atime', 'st_ctime',
'st_gid', 'st_mode', 'st_mtime', 'st_nlink', 'st_size', 'st_uid'))
def certificates(self, conf):
for dentry in os.listdir(conf.root):
if conf.filter_cert(dentry):
fullpath = os.path.join(conf.root, dentry)
if os.path.isdir(fullpath):
yield dentry
def get_conf(self):
return self.configuration
# uid/gid-related helpers; in the end, the xid (uid/gid) precedence is:
# - filespec/same-xid-as
# - filespec/xid
# - conf/same-xid-as
# - conf/xid
# - DEFAULT_XID
def get_uid_gid(self, conf, filespec=None):
"""
Just-get-it-done wrapper around get_{uid,gid}_{global,for_filespec}.
"""
stats = {}
if filespec is None:
return self.get_uid_global(conf, stats), self.get_gid_global(conf, stats)
return self.get_uid_for_filespec(conf, filespec, stats), self.get_gid_for_filespec(conf, filespec, stats)
def get_uid_for_filespec(self, conf, filespec, stats):
"""
File-specific uid selection mechanism: attempt to use file-specific same-uid-as, falling back on
file-specific uid, falling back on global uid selection mechanism.
"""
uid = self.get_stat_attr(filespec.get('same-uid-as', None), 'st_uid', filespec.get('uid', None), stats)
if uid is None:
uid = self.get_uid_global(conf, stats)
return uid
def get_gid_for_filespec(self, conf, filespec, stats):
"""
File-specific gid selection mechanism: attempt to use file-specific same-gid-as, falling back on
file-specific gid, falling back on global gid selection mechanism.
"""
gid = self.get_stat_attr(filespec.get('same-gid-as', None), 'st_gid', filespec.get('gid', None), stats)
if gid is None:
gid = self.get_gid_global(conf, stats)
return gid
def get_uid_global(self, conf, stats):
"""
Global uid selection mechanism: attempt to use same-uid-as, falling back on uid.
"""
return self.get_stat_attr(conf.same_uid_as, 'st_uid', conf.uid, stats)
def get_gid_global(self, conf, stats):
"""
Global gid selection mechanism: attempt to use same-gid-as, falling back on gid.
"""
return self.get_stat_attr(conf.same_gid_as, 'st_gid', conf.gid, stats)
def get_stat_attr(self, path, attr, default, stats):
"""
Stat path and return the request attribute, or the default value if something goes wrong.
"""
if path is None:
return default
try:
return getattr(self.get_stat(path, stats), attr)
except:
return default
def get_stat(self, path, stats):
"""
Simple wrapper around os.stat() that uses a dict to implement some basic caching (for the sake of
uid/gid consistency, not actually for performance). Return either None or a stat structure.
Should throw no exceptions as long as stats is provided.
"""
stat = stats.get(path)
if stat is None:
try:
stats[path] = stat = os.stat(path)
except:
pass
return stat
# End of uid/gid-related helpers
def iterate_paths(self, func, paths):
for filepath in paths:
try:
if type(filepath) is list:
# Array of file paths: look for the first existing path:
for index, subpath in enumerate(filepath):
try:
func(subpath)
# Still there? The file must exist, exit the loop:
break
except OSError as ose:
if ose.errno == errno.ENOENT and index < len(filepath) - 1:
# The file does not exist, try the next one, if any:
continue
else:
# Reached the last file path or encountered another error:
raise
else:
# Presumably a regular file path
func(filepath)
except OSError as ose:
raise FuseOSError(ose.errno)
def register_fd(self, file_descriptor):
with self.filedesc_lock:
self.filedesc_index += 1
new_fd_index = self.filedesc_index
self.filedesc[new_fd_index] = file_descriptor
return new_fd_index
def reload(self, conf=None):
try:
if conf is None:
conf = self.get_conf()
new_conf = CombinedFSConfiguration(conf.path)
# Assuming CPython, this should result in a single STORE_ATTR opcode.
# Since this class features no __setattr__ implementation, the
# resulting execution should be atomic.
self.configuration = new_conf
return new_conf
except:
return None
def handle_reload_getattr(self, conf, fh):
uid, gid = self.get_uid_gid(conf)
return {
'st_nlink': 1,
'st_uid': uid,
'st_gid': gid,
'st_size': RELOAD_FILESIZE,
'st_mode': stat.S_IFREG | conf.key_mode,
}
def handle_reload_open(self, conf, flags):
new_conf = self.reload(conf)
new_fd = {
'conf': conf if new_conf is None else new_conf,
'data': RELOAD_MSG_FAIL if new_conf is None else RELOAD_MSG_OK,
}
return self.register_fd(new_fd)
# Filesystem methods
def access(self, path, mode):
"""
libfuse documentation states:
This will be called for the access() system call. If the
'default_permissions' mount option is given, this method is not called.
Since this program enforces default_permissions, this method will never
be called, which makes it dead simple to implement.
"""
raise FuseOSError(errno.ENOTSUP)
def getattr(self, path, fh=None):
conf = self.get_conf()
if path == RELOAD_PATH:
return self.handle_reload_getattr(conf, fh)
cert, filename, file_spec = conf.analyse_path(path)
if filename is None: # Directory
full_path = os.path.join(conf.root, path.lstrip('/'))
try:
dir_attrs = self.attributes(full_path)
except OSError as ose:
if ose.errno == errno.ENOENT and path == '/':
# Non-existent "live" directory, most likely a misconf,
# fake it to preserve access to RELOAD_PATH:
dir_attrs = {'st_nlink': 2, 'st_size': 4096}
for prop in TIME_PROPS:
dir_attrs[prop] = 0
else:
raise
uid, gid = self.get_uid_gid(conf)
dir_attrs['st_uid'] = uid
dir_attrs['st_gid'] = gid
dir_attrs['st_mode'] = stat.S_IFDIR | conf.dir_mode
return dir_attrs
uid, gid = self.get_uid_gid(conf, file_spec)
attrs = {
'st_nlink': 1,
'st_uid': uid,
'st_gid': gid,
'st_size': 0,
}
def_mode = conf.reg_mode
paths = conf.get_paths(cert, file_spec)
if not paths:
# Virtual empty file:
root_stats = os.stat(conf.root)
for prop in TIME_PROPS:
attrs[prop] = getattr(root_stats, prop)
attrs['st_mode'] = stat.S_IFREG | read_mode_setting(file_spec, 'mode', def_mode)
return attrs
# One array to hold the actual, successive filepaths, one dict to hold
# the latest stat() result for each file:
filepaths = []
stats = {}
def stat_file(path):
stats[path] = os.stat(path)
filepaths.append(path)
self.iterate_paths(stat_file, paths)
for filepath in filepaths:
stat_obj = stats[filepath]
# Pick the highest/latest value for access/change/modification times:
for prop in TIME_PROPS:
prop_val = getattr(stat_obj, prop)
if prop_val > attrs.get(prop, 0):
attrs[prop] = prop_val
# Add up sizes:
attrs['st_size'] += stat_obj.st_size
# Lower permissions if necessary:
if conf.is_sensitive_file(filepath):
def_mode = conf.key_mode
attrs['st_mode'] = stat.S_IFREG | read_mode_setting(file_spec, 'mode', def_mode)
return attrs
def readdir(self, path, fh):
conf = self.get_conf()
cert, filename, _ = conf.analyse_path(path)
# Deal only with directories:
if filename:
raise FuseOSError(errno.ENOTDIR)
# Extra attributes, just what it takes to support dirent->d_type:
dir_attrs = {'st_mode': stat.S_IFDIR }
reg_attrs = {'st_mode': stat.S_IFREG }
# Yield common directory entries:
yield '.', dir_attrs, 0
yield '..', dir_attrs, 0
if not cert:
# Top-level directory
flat_mode = conf.separator != '/'
for cert in self.certificates(conf):
if flat_mode:
for filename in conf.files:
yield cert + conf.separator + filename, reg_attrs, 0
else:
yield cert, dir_attrs, 0
else:
# Second-level directory
for filename in conf.files:
yield filename, reg_attrs, 0
def open(self, path, flags):
conf = self.get_conf()
if path == RELOAD_PATH:
return self.handle_reload_open(conf, flags)
cert, filename, file_spec = conf.analyse_path(path)
if not cert or not filename:
raise FuseOSError(errno.ENOENT)
# Being a read-only filesystem spares us the need to check most flags.
new_fd = {
'conf': conf,
'cert': cert,
'filename': filename,
'file_spec': file_spec,
}
return self.register_fd(new_fd)
def read(self, path, length, offset, fh):
filedesc = self.filedesc.get(fh)
# Use the same configuration as open() when it created the file descriptor:
conf = filedesc['conf']
if filedesc is None:
raise FuseOSError(errno.EBADF)
data = filedesc.get('data')
if data is None:
paths = conf.get_paths(filedesc['cert'], filedesc['file_spec'])
data = {'data': bytes() }
def concatenate(path):
data['data'] += open(path, 'rb').read()
self.iterate_paths(concatenate, paths)
filedesc['data'] = data = data['data']
read_chunk = data[offset:offset + length]
return read_chunk
def release(self, path, fh):
with self.filedesc_lock:
del self.filedesc[fh]
# Integers in Python have arbitrary precision, i.e. they are unbounded
# and thus exempt from overflows as long as they are manipulated in
# pure Python.
def statfs(self, path):
stv = os.statvfs(self.get_conf().root)
return dict((key, getattr(stv, key)) for key in ('f_bavail', 'f_bfree',
'f_blocks', 'f_bsize', 'f_favail', 'f_ffree', 'f_files', 'f_flag',
'f_frsize', 'f_namemax'))
def readlink(self, path):
# We never expose any symlink, therefore it should be safe to always
# return EINVAL:
raise FuseOSError(errno.EINVAL)
def main(conf_path, mountpoint, foreground):
FUSE(CombinedFS(conf_path), mountpoint, foreground=foreground, ro=True, default_permissions=True, allow_other=True)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Expose a transformed, version of Let\'s Encrypt / Certbot\'s "live" directory')
parser.add_argument('conf_path', help='CombinedFS configuration file')
parser.add_argument('mountpoint', help='mount point')
parser.add_argument('-o', dest='options', help='mount options (ignored, only there for compatibility purposes)')
parser.add_argument('-f', '--foreground', dest='foreground', help='run in the foreground', action='store_true')
args = parser.parse_args()
main(args.conf_path, args.mountpoint, args.foreground)