#!/usr/bin/python # #Copyright (c) 2010 Cloud Linux LTD #All rights reserved. # #Redistribution and use in source and binary forms, with or without #modification, are permitted provided that the following conditions #are met: # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above # copyright notice, this list of conditions and the following # disclaimer in the documentation and/or other materials provided # with the distribution. # * The names of its contributors may not be used to endorse or # promote products derived from this software without specific # prior written permission. # #THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS #"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT #LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS #FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE #COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, #INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, #BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; #LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER #CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT #LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN #ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE #POSSIBILITY OF SUCH DAMAGE. # # This module provides functions for secure I/O and filesystem operations import os from ctypes import * libc = cdll.LoadLibrary("libc.so.6") liblve = cdll.LoadLibrary("liblve.so.0") # --- libc functions ----------------------------------------------- ino_t = c_long off_t = c_long class DIRENTRY(Structure): _fields_ = [ ('d_ino', ino_t), # inode number ('d_off', off_t), # offset to the next dirent ('d_reclen', c_ushort), # length of this record ('d_type', c_ubyte), # type of file; not supported by all file system types ('d_name', c_char * 256), # filename ] DIRENTRY_P = POINTER(DIRENTRY) fchown = libc.fchown fchown.argtypes = [c_int, c_int, c_int] fchown.restype = c_int fchmod = libc.fchmod fchmod.argtypes = [c_int, c_int] fchmod.restype = c_int fdopendir = libc.fdopendir # accepts file/dir descriptor (integer) fdopendir.argtypes = [c_int] # returns pointer to DIR structure fdopendir.restype = c_void_p readdir = libc.readdir # accepts pointer to DIR structure readdir.argtypes = [c_void_p] # returns pointer to DIRENTRY structure readdir.restype = DIRENTRY_P rewinddir = libc.rewinddir # accepts pointer to DIR structure rewinddir.argtypes = [c_void_p] # returns void rewinddir.restype = None closedir = libc.closedir # accepts pointer to DIR structure closedir.argtypes = [c_void_p] closedir.restype = c_int # --- liblve functions ----------------------------------------------- # Opens path for reading not following symlinks and verifies that opened path is inside parent_path # Returns: # descriptor if successful # -1 if path does not exist or is a symlink # -2 if opened path is NOT inside parent_path or cannot be determined # accepts path, parent_path open_not_symlink = liblve.open_not_symlink open_not_symlink.argtypes = [c_char_p, c_char_p] open_not_symlink.restype = c_int # Closes descriptor (if it is not equal -1) # accepts file/dir descriptor (integer) liblve.closefd.argtypes = [c_int] # returns void liblve.closefd.restype = None # Tries to read first directory entry in order to ensure that descriptor is valid # Returns 0 if reading succeeded or -1 if error has occured check_dir = liblve.check_dir # accepts descriptor check_dir.argtypes = [c_int] check_dir.restype = c_int # Checks if path is a directory (in secure manner) # Also opens path (if descriptor fd == -1) and then checks that opened path is inside parent_path # Returns descriptor if path refers to directory # Returns -1 if path does not exist or is not a directory # Returns -2 if opened path is NOT inside parent_path or cannot be determined isdir = liblve.isdir # accepts path, descriptor, parent_path isdir.argtypes = [c_char_p, c_int, c_char_p] isdir.restype = c_int # Sets permissions to directory (in secure manner) # Returns descriptor if successful # Returns -1 if error has occured # Returns -2 if opened path is NOT inside parent_path or cannot be determined # accepts: const char *path, mode_t perm, int fd, const char *parent_path liblve.set_perm_dir_secure.argtypes = [c_char_p, c_int, c_int, c_char_p] liblve.set_perm_dir_secure.restype = c_int # Sets owner and group of directory (in secure manner) # Returns descriptor if successful # Returns -1 if error has occured # Returns -2 if opened path is NOT inside parent_path or cannot be determined # accepts: const char *path, uid_t uid, gid_t gid, int fd, const char *parent_path liblve.set_owner_dir_secure.argtypes = [c_char_p, c_int, c_int, c_int, c_char_p] liblve.set_owner_dir_secure.restype = c_int # Creates directory if it does not exist, sets permissions/owner otherwise # Returns descriptor if successful # Returns -1 if error has occured # accepts: const char *path, mode_t perm, uid_t uid, gid_t gid, int fd, const char *parent_path liblve.create_dir_secure.argtypes = [c_char_p, c_int, c_int, c_int, c_int, c_char_p] liblve.create_dir_secure.restype = c_int # Recursive directory creation function # Returns 0 if successful # Returns -1 if error has occured # accepts: const char *path, mode_t perm, uid_t uid, gid_t gid, const char *parent_path liblve.makedirs_secure.argtypes = [c_char_p, c_int, c_int, c_int, c_char_p] liblve.makedirs_secure.restype = c_int # Writes absolute path pointed by descriptor fd to buffer *buf # Returns buf if successful # Returns NULL if error has occured get_path_from_descriptor = liblve.get_path_from_descriptor get_path_from_descriptor.argtypes = [c_int, c_char_p] get_path_from_descriptor.restype = c_char_p # Returns 1 if subdir is subdirectory of dir, 0 otherwise is_subdir = liblve.is_subdir is_subdir.argtypes = [c_char_p, c_char_p] is_subdir.restype = c_int # Returns list of entries of directory pointed by descriptor def flistdir(fd): # Duplicate descriptor, because closedir() closes descriptor associated with directory stream fd2 = os.dup(fd) # Open directory stream dirp = fdopendir(fd2) if not dirp: raise Exception("fdopendir error") # Reset position of directory stream # (so it will be possible to read content of directory multiple times # via other descriptors that refer to the directory) rewinddir(dirp) dirlist = [] while True: entryp = readdir(dirp) if not entryp: break entry = entryp.contents dirlist.append(entry.d_name) rewinddir(dirp) closedir(dirp) return dirlist def closefd(fd): if fd is not None: try: os.close(fd) except OSError: pass # Sets permissions to directory (in secure manner) # Returns descriptor if successful # Returns None if error has occured def set_perm_dir_secure(path, perm, parent_path, fd = None, logger = None): if fd == None: fd = -1 fd = liblve.set_perm_dir_secure(path, perm, fd, parent_path) if fd > 0: return fd if logger is not None: logger('Error: failed to set permissions of directory ' + path, False, True) return None # Sets owner and group of directory (in secure manner) # Returns descriptor if successful # Returns None if error has occured def set_owner_dir_secure(path, uid, gid, parent_path, fd = None, logger = None): if fd == None: fd = -1 fd = liblve.set_owner_dir_secure(path, uid, gid, fd, parent_path) if fd > 0: return fd if logger is not None: logger('Error: failed to set owner of directory ' + path, False, True) return None # Creates directory if it does not exist, sets permissions/owner otherwise # Returns descriptor if successful # Returns None if error has occured def create_dir_secure(path, perm, uid, gid, parent_path, fd = None, logger = None): if fd == None: fd = -1 fd = liblve.create_dir_secure(path, perm, uid, gid, fd, parent_path) if fd > 0: return fd if logger is not None: logger('Error: failed to create directory ' + path, False, True) return None # Recursive directory creation function # Returns 0 if successful # Returns -1 if error has occured def makedirs_secure(path, perm, uid, gid, parent_path, logger = None): res = liblve.makedirs_secure(path, perm, uid, gid, parent_path) if res and logger: logger('Error: failed to create directory ' + path, False, True) return res def set_capability(clear=False): """ Set CAP_SYS_RESOURCE capability :param bool clear: Set on if it's true, set off otherwise :return: 0 for success, -1 otherwise :rtype: int """ if clear: return liblve.disable_quota_capability() else: return liblve.enable_quota_capability() def change_uid(uid): """ Change effective uid of current process and set CAP_SYS_RESOURCE capbality to prevent "Disk quota exceeded" error :param int euid: User ID to set it as current effective UID :return: 0 if capability was set successfuly, -1 otherwise :rtype: int """ os.setuid(uid) return set_capability()