blob: 2bc9bc1dbf6c8750b7bf63f94e9f15575ea15624 [file] [log] [blame]
#!/usr/bin/env python
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Stub implementations of restricted functions."""
import errno
import functools
import inspect
import locale
import mimetypes
import os
import random
import re
import sys
import threading
# sysconfig is new in Python 2.7.
try:
import sysconfig
except ImportError:
sysconfig = None
def os_error_not_implemented(*unused_args, **unused_kwargs):
raise OSError(errno.ENOSYS, 'Function not implemented')
def return_minus_one(*unused_args, **unused_kwargs):
return -1
def fake_uname():
"""Fake version of os.uname."""
return ('Linux', '', '', '', '')
def fake_urandom(n):
"""Fake version of os.urandom."""
# On Mac OS X, os.urandom reads /dev/urandom from Python code, which is
# disallowed by the sandbox.
return ''.join(chr(random.randint(0, 255)) for _ in xrange(n))
def fake_access(path, mode):
"""Fake version of os.access where only reads are supported."""
if mode & (os.W_OK | os.X_OK):
return False
else:
return FakeFile.is_file_accessible(path)
def fake_open(filename, flags, mode=0777, _os_open=os.open):
"""Fake version of os.open."""
# A copy of os.open is saved in _os_open so it can still be used after os.open
# is replaced with this stub.
if flags & (os.O_RDWR | os.O_CREAT | os.O_WRONLY):
raise OSError(errno.EROFS, 'Read-only file system', filename)
elif not FakeFile.is_file_accessible(filename):
raise OSError(errno.ENOENT, 'No such file or directory', filename)
return _os_open(filename, flags, mode)
def fake_set_locale(category, value=None, original_setlocale=locale.setlocale):
"""Fake version of locale.setlocale that only supports the default."""
if value not in (None, '', 'C', 'POSIX'):
raise locale.Error('locale emulation only supports "C" locale')
return original_setlocale(category, 'C')
def fake_get_platform():
"""Fake distutils.util.get_platform()."""
if sys.platform == 'darwin':
return 'macosx-'
else:
return 'linux-'
class FakeFile(file):
"""File sub-class that enforces the restrictions of production."""
ALLOWED_MODES = frozenset(['r', 'rb', 'U', 'rU'])
# Individual files that are allowed to be accessed.
ALLOWED_FILES = set(os.path.normcase(filename)
for filename in mimetypes.knownfiles
if os.path.isfile(filename))
# Directories which are allowed to be accessed. All sub-directories are
# also allowed.
ALLOWED_DIRS = set([
os.path.normcase(os.path.realpath(os.path.dirname(os.__file__))),
os.path.normcase(os.path.abspath(os.path.dirname(os.__file__))),
os.path.normcase(os.path.dirname(os.path.realpath(os.__file__))),
os.path.normcase(os.path.dirname(os.path.abspath(os.__file__))),
])
os_source_location = inspect.getsourcefile(os)
# inspect.getsource may return None if it cannot find the os.py file.
if os_source_location is not None:
# Whitelist source file location to allow virtualenv to work.
# This is necessary because the compiled bytecode may be in a different
# location to the source code.
ALLOWED_DIRS.update([
os.path.normcase(os.path.realpath(os.path.dirname(os_source_location))),
os.path.normcase(os.path.abspath(os.path.dirname(os_source_location))),
os.path.normcase(os.path.dirname(os.path.realpath(os_source_location))),
os.path.normcase(os.path.dirname(os.path.abspath(os_source_location))),
])
# sysconfig requires access to config.h.
if sysconfig:
ALLOWED_DIRS.add(os.path.dirname(sysconfig.get_config_h_filename()))
# Configuration - set_allowed_paths must be called to initialize them.
_allowed_dirs = None # List of accessible paths.
# Configuration - set_skip_files must be called to initialize it.
_skip_files = None # Regex of skip files.
# Configuration - set_static_files must be called to initialize it.
_static_files = None # Regex of static files.
# Cache for results of is_file_accessible, {absolute filename: Boolean}
_availability_cache = {}
_availability_cache_lock = threading.Lock()
@staticmethod
def set_allowed_paths(root_path, application_paths):
"""Configures which paths are allowed to be accessed.
Must be called at least once before any file objects are created in the
hardened environment.
Args:
root_path: Absolute path to the root of the application.
application_paths: List of additional paths that the application may
access, this must include the App Engine runtime but not the Python
library directories.
"""
# Use os.path.realpath to flush out symlink-at-root issues.
# (Deeper symlinks will not use realpath.)
_application_paths = (set(os.path.realpath(path)
for path in application_paths) |
set(os.path.abspath(path)
for path in application_paths))
FakeFile._root_path = os.path.normcase(os.path.abspath(root_path))
_application_paths.add(FakeFile._root_path)
FakeFile._allowed_dirs = _application_paths | FakeFile.ALLOWED_DIRS
with FakeFile._availability_cache_lock:
FakeFile._availability_cache = {}
@staticmethod
def set_skip_files(skip_files):
"""Configure the skip_files regex.
Files that match this regex are inaccessible in the hardened environment.
Must be called at least once before any file objects are created in the
hardened environment.
Args:
skip_files: A str containing a regex to match against file paths.
"""
FakeFile._skip_files = re.compile(skip_files)
with FakeFile._availability_cache_lock:
FakeFile._availability_cache = {}
@staticmethod
def set_static_files(static_files):
"""Configure the static_files regex.
Files that match this regex are inaccessible in the hardened environment.
Must be called at least once before any file objects are created in the
hardened environment.
Args:
static_files: A str containing a regex to match against file paths.
"""
FakeFile._static_files = re.compile(static_files)
with FakeFile._availability_cache_lock:
FakeFile._availability_cache = {}
@staticmethod
def is_file_accessible(filename):
"""Determines if a file is accessible.
set_allowed_paths(), set_skip_files() and SetStaticFileConfigMatcher() must
be called before this method or else all file accesses will raise an error.
Args:
filename: Path of the file to check (relative or absolute). May be a
directory, in which case access for files inside that directory will
be checked.
Returns:
True if the file is accessible, False otherwise.
Raises:
TypeError: filename is not a basestring.
"""
if not isinstance(filename, basestring):
raise TypeError()
# Blindly follow symlinks here. DO NOT use os.path.realpath. This approach
# enables the application developer to create symlinks in their
# application directory to any additional libraries they want to use.
fixed_filename = os.path.normcase(os.path.abspath(filename))
# Some of the access checks are expensive as they look through a list of
# regular expressions for a match. As long as app.yaml isn't changed the
# answer will always be the same as it's only depending on the filename and
# the configuration, not on which files do exist or not.
with FakeFile._availability_cache_lock:
result = FakeFile._availability_cache.get(fixed_filename)
if result is None:
if (_is_path_in_directories(fixed_filename, [FakeFile._root_path]) and
fixed_filename != FakeFile._root_path):
relative_filename = fixed_filename[len(FakeFile._root_path):].lstrip(
os.path.sep)
block_access = (FakeFile._skip_files.match(relative_filename) or
FakeFile._static_files.match(relative_filename))
else:
block_access = False
result = not block_access and (
fixed_filename in FakeFile.ALLOWED_FILES or
_is_path_in_directories(fixed_filename, FakeFile._allowed_dirs))
with FakeFile._availability_cache_lock:
FakeFile._availability_cache[fixed_filename] = result
return result
def __init__(self, filename, mode='r', bufsize=-1, **kwargs):
"""Initializer. See file built-in documentation."""
if mode not in FakeFile.ALLOWED_MODES:
raise IOError(errno.EROFS, 'Read-only file system', filename)
if not FakeFile.is_file_accessible(filename):
raise IOError(errno.EACCES, 'file not accessible', filename)
super(FakeFile, self).__init__(filename, mode, bufsize, **kwargs)
class RestrictedPathFunction(object):
"""Enforces restrictions for functions with a path as their first argument."""
def __init__(self, original_func):
"""Initializer.
Args:
original_func: Callable that takes as its first argument the path to a
file or directory on disk; all subsequent arguments may be variable.
"""
self._original_func = original_func
functools.update_wrapper(self, original_func)
def __call__(self, path, *args, **kwargs):
"""Enforces access permissions for the wrapped function."""
if not FakeFile.is_file_accessible(path):
raise OSError(errno.EACCES, 'path not accessible', path)
return self._original_func(path, *args, **kwargs)
def _is_path_in_directories(filename, directories):
"""Determines if a filename is contained within one of a set of directories.
Args:
filename: Path of the file (relative or absolute).
directories: Iterable collection of paths to directories which the
given filename may be under.
Returns:
True if the supplied filename is in one of the given sub-directories or
its hierarchy of children. False otherwise.
"""
fixed_path = os.path.normcase(os.path.abspath(filename))
for parent in directories:
fixed_parent = os.path.normcase(os.path.abspath(parent))
if os.path.commonprefix([fixed_path, fixed_parent]) == fixed_parent:
return True
return False