437 lines
15 KiB
Python
437 lines
15 KiB
Python
|
"""A sandbox layer that ensures unsafe operations cannot be performed.
|
||
|
|
Useful when the template itself comes from an untrusted source.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import operator
|
||
|
|
import types
|
||
|
|
import typing as t
|
||
|
|
from _string import formatter_field_name_split # type: ignore
|
||
|
|
from collections import abc
|
||
|
|
from collections import deque
|
||
|
|
from functools import update_wrapper
|
||
|
|
from string import Formatter
|
||
|
|
|
||
|
|
from markupsafe import EscapeFormatter
|
||
|
|
from markupsafe import Markup
|
||
|
|
|
||
|
|
from .environment import Environment
|
||
|
|
from .exceptions import SecurityError
|
||
|
|
from .runtime import Context
|
||
|
|
from .runtime import Undefined
|
||
|
|
|
||
|
|
F = t.TypeVar("F", bound=t.Callable[..., t.Any])
|
||
|
|
|
||
|
|
#: maximum number of items a range may produce
|
||
|
|
MAX_RANGE = 100000
|
||
|
|
|
||
|
|
#: Unsafe function attributes.
|
||
|
|
UNSAFE_FUNCTION_ATTRIBUTES: t.Set[str] = set()
|
||
|
|
|
||
|
|
#: Unsafe method attributes. Function attributes are unsafe for methods too.
|
||
|
|
UNSAFE_METHOD_ATTRIBUTES: t.Set[str] = set()
|
||
|
|
|
||
|
|
#: unsafe generator attributes.
|
||
|
|
UNSAFE_GENERATOR_ATTRIBUTES = {"gi_frame", "gi_code"}
|
||
|
|
|
||
|
|
#: unsafe attributes on coroutines
|
||
|
|
UNSAFE_COROUTINE_ATTRIBUTES = {"cr_frame", "cr_code"}
|
||
|
|
|
||
|
|
#: unsafe attributes on async generators
|
||
|
|
UNSAFE_ASYNC_GENERATOR_ATTRIBUTES = {"ag_code", "ag_frame"}
|
||
|
|
|
||
|
|
_mutable_spec: t.Tuple[t.Tuple[t.Type[t.Any], t.FrozenSet[str]], ...] = (
|
||
|
|
(
|
||
|
|
abc.MutableSet,
|
||
|
|
frozenset(
|
||
|
|
[
|
||
|
|
"add",
|
||
|
|
"clear",
|
||
|
|
"difference_update",
|
||
|
|
"discard",
|
||
|
|
"pop",
|
||
|
|
"remove",
|
||
|
|
"symmetric_difference_update",
|
||
|
|
"update",
|
||
|
|
]
|
||
|
|
),
|
||
|
|
),
|
||
|
|
(
|
||
|
|
abc.MutableMapping,
|
||
|
|
frozenset(["clear", "pop", "popitem", "setdefault", "update"]),
|
||
|
|
),
|
||
|
|
(
|
||
|
|
abc.MutableSequence,
|
||
|
|
frozenset(
|
||
|
|
["append", "clear", "pop", "reverse", "insert", "sort", "extend", "remove"]
|
||
|
|
),
|
||
|
|
),
|
||
|
|
(
|
||
|
|
deque,
|
||
|
|
frozenset(
|
||
|
|
[
|
||
|
|
"append",
|
||
|
|
"appendleft",
|
||
|
|
"clear",
|
||
|
|
"extend",
|
||
|
|
"extendleft",
|
||
|
|
"pop",
|
||
|
|
"popleft",
|
||
|
|
"remove",
|
||
|
|
"rotate",
|
||
|
|
]
|
||
|
|
),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def safe_range(*args: int) -> range:
|
||
|
|
"""A range that can't generate ranges with a length of more than
|
||
|
|
MAX_RANGE items.
|
||
|
|
"""
|
||
|
|
rng = range(*args)
|
||
|
|
|
||
|
|
if len(rng) > MAX_RANGE:
|
||
|
|
raise OverflowError(
|
||
|
|
"Range too big. The sandbox blocks ranges larger than"
|
||
|
|
f" MAX_RANGE ({MAX_RANGE})."
|
||
|
|
)
|
||
|
|
|
||
|
|
return rng
|
||
|
|
|
||
|
|
|
||
|
|
def unsafe(f: F) -> F:
|
||
|
|
"""Marks a function or method as unsafe.
|
||
|
|
|
||
|
|
.. code-block: python
|
||
|
|
|
||
|
|
@unsafe
|
||
|
|
def delete(self):
|
||
|
|
pass
|
||
|
|
"""
|
||
|
|
f.unsafe_callable = True # type: ignore
|
||
|
|
return f
|
||
|
|
|
||
|
|
|
||
|
|
def is_internal_attribute(obj: t.Any, attr: str) -> bool:
|
||
|
|
"""Test if the attribute given is an internal python attribute. For
|
||
|
|
example this function returns `True` for the `func_code` attribute of
|
||
|
|
python objects. This is useful if the environment method
|
||
|
|
:meth:`~SandboxedEnvironment.is_safe_attribute` is overridden.
|
||
|
|
|
||
|
|
>>> from jinja2.sandbox import is_internal_attribute
|
||
|
|
>>> is_internal_attribute(str, "mro")
|
||
|
|
True
|
||
|
|
>>> is_internal_attribute(str, "upper")
|
||
|
|
False
|
||
|
|
"""
|
||
|
|
if isinstance(obj, types.FunctionType):
|
||
|
|
if attr in UNSAFE_FUNCTION_ATTRIBUTES:
|
||
|
|
return True
|
||
|
|
elif isinstance(obj, types.MethodType):
|
||
|
|
if attr in UNSAFE_FUNCTION_ATTRIBUTES or attr in UNSAFE_METHOD_ATTRIBUTES:
|
||
|
|
return True
|
||
|
|
elif isinstance(obj, type):
|
||
|
|
if attr == "mro":
|
||
|
|
return True
|
||
|
|
elif isinstance(obj, (types.CodeType, types.TracebackType, types.FrameType)):
|
||
|
|
return True
|
||
|
|
elif isinstance(obj, types.GeneratorType):
|
||
|
|
if attr in UNSAFE_GENERATOR_ATTRIBUTES:
|
||
|
|
return True
|
||
|
|
elif hasattr(types, "CoroutineType") and isinstance(obj, types.CoroutineType):
|
||
|
|
if attr in UNSAFE_COROUTINE_ATTRIBUTES:
|
||
|
|
return True
|
||
|
|
elif hasattr(types, "AsyncGeneratorType") and isinstance(
|
||
|
|
obj, types.AsyncGeneratorType
|
||
|
|
):
|
||
|
|
if attr in UNSAFE_ASYNC_GENERATOR_ATTRIBUTES:
|
||
|
|
return True
|
||
|
|
return attr.startswith("__")
|
||
|
|
|
||
|
|
|
||
|
|
def modifies_known_mutable(obj: t.Any, attr: str) -> bool:
|
||
|
|
"""This function checks if an attribute on a builtin mutable object
|
||
|
|
(list, dict, set or deque) or the corresponding ABCs would modify it
|
||
|
|
if called.
|
||
|
|
|
||
|
|
>>> modifies_known_mutable({}, "clear")
|
||
|
|
True
|
||
|
|
>>> modifies_known_mutable({}, "keys")
|
||
|
|
False
|
||
|
|
>>> modifies_known_mutable([], "append")
|
||
|
|
True
|
||
|
|
>>> modifies_known_mutable([], "index")
|
||
|
|
False
|
||
|
|
|
||
|
|
If called with an unsupported object, ``False`` is returned.
|
||
|
|
|
||
|
|
>>> modifies_known_mutable("foo", "upper")
|
||
|
|
False
|
||
|
|
"""
|
||
|
|
for typespec, unsafe in _mutable_spec:
|
||
|
|
if isinstance(obj, typespec):
|
||
|
|
return attr in unsafe
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
class SandboxedEnvironment(Environment):
|
||
|
|
"""The sandboxed environment. It works like the regular environment but
|
||
|
|
tells the compiler to generate sandboxed code. Additionally subclasses of
|
||
|
|
this environment may override the methods that tell the runtime what
|
||
|
|
attributes or functions are safe to access.
|
||
|
|
|
||
|
|
If the template tries to access insecure code a :exc:`SecurityError` is
|
||
|
|
raised. However also other exceptions may occur during the rendering so
|
||
|
|
the caller has to ensure that all exceptions are caught.
|
||
|
|
"""
|
||
|
|
|
||
|
|
sandboxed = True
|
||
|
|
|
||
|
|
#: default callback table for the binary operators. A copy of this is
|
||
|
|
#: available on each instance of a sandboxed environment as
|
||
|
|
#: :attr:`binop_table`
|
||
|
|
default_binop_table: t.Dict[str, t.Callable[[t.Any, t.Any], t.Any]] = {
|
||
|
|
"+": operator.add,
|
||
|
|
"-": operator.sub,
|
||
|
|
"*": operator.mul,
|
||
|
|
"/": operator.truediv,
|
||
|
|
"//": operator.floordiv,
|
||
|
|
"**": operator.pow,
|
||
|
|
"%": operator.mod,
|
||
|
|
}
|
||
|
|
|
||
|
|
#: default callback table for the unary operators. A copy of this is
|
||
|
|
#: available on each instance of a sandboxed environment as
|
||
|
|
#: :attr:`unop_table`
|
||
|
|
default_unop_table: t.Dict[str, t.Callable[[t.Any], t.Any]] = {
|
||
|
|
"+": operator.pos,
|
||
|
|
"-": operator.neg,
|
||
|
|
}
|
||
|
|
|
||
|
|
#: a set of binary operators that should be intercepted. Each operator
|
||
|
|
#: that is added to this set (empty by default) is delegated to the
|
||
|
|
#: :meth:`call_binop` method that will perform the operator. The default
|
||
|
|
#: operator callback is specified by :attr:`binop_table`.
|
||
|
|
#:
|
||
|
|
#: The following binary operators are interceptable:
|
||
|
|
#: ``//``, ``%``, ``+``, ``*``, ``-``, ``/``, and ``**``
|
||
|
|
#:
|
||
|
|
#: The default operation form the operator table corresponds to the
|
||
|
|
#: builtin function. Intercepted calls are always slower than the native
|
||
|
|
#: operator call, so make sure only to intercept the ones you are
|
||
|
|
#: interested in.
|
||
|
|
#:
|
||
|
|
#: .. versionadded:: 2.6
|
||
|
|
intercepted_binops: t.FrozenSet[str] = frozenset()
|
||
|
|
|
||
|
|
#: a set of unary operators that should be intercepted. Each operator
|
||
|
|
#: that is added to this set (empty by default) is delegated to the
|
||
|
|
#: :meth:`call_unop` method that will perform the operator. The default
|
||
|
|
#: operator callback is specified by :attr:`unop_table`.
|
||
|
|
#:
|
||
|
|
#: The following unary operators are interceptable: ``+``, ``-``
|
||
|
|
#:
|
||
|
|
#: The default operation form the operator table corresponds to the
|
||
|
|
#: builtin function. Intercepted calls are always slower than the native
|
||
|
|
#: operator call, so make sure only to intercept the ones you are
|
||
|
|
#: interested in.
|
||
|
|
#:
|
||
|
|
#: .. versionadded:: 2.6
|
||
|
|
intercepted_unops: t.FrozenSet[str] = frozenset()
|
||
|
|
|
||
|
|
def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
|
||
|
|
super().__init__(*args, **kwargs)
|
||
|
|
self.globals["range"] = safe_range
|
||
|
|
self.binop_table = self.default_binop_table.copy()
|
||
|
|
self.unop_table = self.default_unop_table.copy()
|
||
|
|
|
||
|
|
def is_safe_attribute(self, obj: t.Any, attr: str, value: t.Any) -> bool:
|
||
|
|
"""The sandboxed environment will call this method to check if the
|
||
|
|
attribute of an object is safe to access. Per default all attributes
|
||
|
|
starting with an underscore are considered private as well as the
|
||
|
|
special attributes of internal python objects as returned by the
|
||
|
|
:func:`is_internal_attribute` function.
|
||
|
|
"""
|
||
|
|
return not (attr.startswith("_") or is_internal_attribute(obj, attr))
|
||
|
|
|
||
|
|
def is_safe_callable(self, obj: t.Any) -> bool:
|
||
|
|
"""Check if an object is safely callable. By default callables
|
||
|
|
are considered safe unless decorated with :func:`unsafe`.
|
||
|
|
|
||
|
|
This also recognizes the Django convention of setting
|
||
|
|
``func.alters_data = True``.
|
||
|
|
"""
|
||
|
|
return not (
|
||
|
|
getattr(obj, "unsafe_callable", False) or getattr(obj, "alters_data", False)
|
||
|
|
)
|
||
|
|
|
||
|
|
def call_binop(
|
||
|
|
self, context: Context, operator: str, left: t.Any, right: t.Any
|
||
|
|
) -> t.Any:
|
||
|
|
"""For intercepted binary operator calls (:meth:`intercepted_binops`)
|
||
|
|
this function is executed instead of the builtin operator. This can
|
||
|
|
be used to fine tune the behavior of certain operators.
|
||
|
|
|
||
|
|
.. versionadded:: 2.6
|
||
|
|
"""
|
||
|
|
return self.binop_table[operator](left, right)
|
||
|
|
|
||
|
|
def call_unop(self, context: Context, operator: str, arg: t.Any) -> t.Any:
|
||
|
|
"""For intercepted unary operator calls (:meth:`intercepted_unops`)
|
||
|
|
this function is executed instead of the builtin operator. This can
|
||
|
|
be used to fine tune the behavior of certain operators.
|
||
|
|
|
||
|
|
.. versionadded:: 2.6
|
||
|
|
"""
|
||
|
|
return self.unop_table[operator](arg)
|
||
|
|
|
||
|
|
def getitem(
|
||
|
|
self, obj: t.Any, argument: t.Union[str, t.Any]
|
||
|
|
) -> t.Union[t.Any, Undefined]:
|
||
|
|
"""Subscribe an object from sandboxed code."""
|
||
|
|
try:
|
||
|
|
return obj[argument]
|
||
|
|
except (TypeError, LookupError):
|
||
|
|
if isinstance(argument, str):
|
||
|
|
try:
|
||
|
|
attr = str(argument)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
try:
|
||
|
|
value = getattr(obj, attr)
|
||
|
|
except AttributeError:
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
fmt = self.wrap_str_format(value)
|
||
|
|
if fmt is not None:
|
||
|
|
return fmt
|
||
|
|
if self.is_safe_attribute(obj, argument, value):
|
||
|
|
return value
|
||
|
|
return self.unsafe_undefined(obj, argument)
|
||
|
|
return self.undefined(obj=obj, name=argument)
|
||
|
|
|
||
|
|
def getattr(self, obj: t.Any, attribute: str) -> t.Union[t.Any, Undefined]:
|
||
|
|
"""Subscribe an object from sandboxed code and prefer the
|
||
|
|
attribute. The attribute passed *must* be a bytestring.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
value = getattr(obj, attribute)
|
||
|
|
except AttributeError:
|
||
|
|
try:
|
||
|
|
return obj[attribute]
|
||
|
|
except (TypeError, LookupError):
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
fmt = self.wrap_str_format(value)
|
||
|
|
if fmt is not None:
|
||
|
|
return fmt
|
||
|
|
if self.is_safe_attribute(obj, attribute, value):
|
||
|
|
return value
|
||
|
|
return self.unsafe_undefined(obj, attribute)
|
||
|
|
return self.undefined(obj=obj, name=attribute)
|
||
|
|
|
||
|
|
def unsafe_undefined(self, obj: t.Any, attribute: str) -> Undefined:
|
||
|
|
"""Return an undefined object for unsafe attributes."""
|
||
|
|
return self.undefined(
|
||
|
|
f"access to attribute {attribute!r} of"
|
||
|
|
f" {type(obj).__name__!r} object is unsafe.",
|
||
|
|
name=attribute,
|
||
|
|
obj=obj,
|
||
|
|
exc=SecurityError,
|
||
|
|
)
|
||
|
|
|
||
|
|
def wrap_str_format(self, value: t.Any) -> t.Optional[t.Callable[..., str]]:
|
||
|
|
"""If the given value is a ``str.format`` or ``str.format_map`` method,
|
||
|
|
return a new function than handles sandboxing. This is done at access
|
||
|
|
rather than in :meth:`call`, so that calls made without ``call`` are
|
||
|
|
also sandboxed.
|
||
|
|
"""
|
||
|
|
if not isinstance(
|
||
|
|
value, (types.MethodType, types.BuiltinMethodType)
|
||
|
|
) or value.__name__ not in ("format", "format_map"):
|
||
|
|
return None
|
||
|
|
|
||
|
|
f_self: t.Any = value.__self__
|
||
|
|
|
||
|
|
if not isinstance(f_self, str):
|
||
|
|
return None
|
||
|
|
|
||
|
|
str_type: t.Type[str] = type(f_self)
|
||
|
|
is_format_map = value.__name__ == "format_map"
|
||
|
|
formatter: SandboxedFormatter
|
||
|
|
|
||
|
|
if isinstance(f_self, Markup):
|
||
|
|
formatter = SandboxedEscapeFormatter(self, escape=f_self.escape)
|
||
|
|
else:
|
||
|
|
formatter = SandboxedFormatter(self)
|
||
|
|
|
||
|
|
vformat = formatter.vformat
|
||
|
|
|
||
|
|
def wrapper(*args: t.Any, **kwargs: t.Any) -> str:
|
||
|
|
if is_format_map:
|
||
|
|
if kwargs:
|
||
|
|
raise TypeError("format_map() takes no keyword arguments")
|
||
|
|
|
||
|
|
if len(args) != 1:
|
||
|
|
raise TypeError(
|
||
|
|
f"format_map() takes exactly one argument ({len(args)} given)"
|
||
|
|
)
|
||
|
|
|
||
|
|
kwargs = args[0]
|
||
|
|
args = ()
|
||
|
|
|
||
|
|
return str_type(vformat(f_self, args, kwargs))
|
||
|
|
|
||
|
|
return update_wrapper(wrapper, value)
|
||
|
|
|
||
|
|
def call(
|
||
|
|
__self, # noqa: B902
|
||
|
|
__context: Context,
|
||
|
|
__obj: t.Any,
|
||
|
|
*args: t.Any,
|
||
|
|
**kwargs: t.Any,
|
||
|
|
) -> t.Any:
|
||
|
|
"""Call an object from sandboxed code."""
|
||
|
|
|
||
|
|
# the double prefixes are to avoid double keyword argument
|
||
|
|
# errors when proxying the call.
|
||
|
|
if not __self.is_safe_callable(__obj):
|
||
|
|
raise SecurityError(f"{__obj!r} is not safely callable")
|
||
|
|
return __context.call(__obj, *args, **kwargs)
|
||
|
|
|
||
|
|
|
||
|
|
class ImmutableSandboxedEnvironment(SandboxedEnvironment):
|
||
|
|
"""Works exactly like the regular `SandboxedEnvironment` but does not
|
||
|
|
permit modifications on the builtin mutable objects `list`, `set`, and
|
||
|
|
`dict` by using the :func:`modifies_known_mutable` function.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def is_safe_attribute(self, obj: t.Any, attr: str, value: t.Any) -> bool:
|
||
|
|
if not super().is_safe_attribute(obj, attr, value):
|
||
|
|
return False
|
||
|
|
|
||
|
|
return not modifies_known_mutable(obj, attr)
|
||
|
|
|
||
|
|
|
||
|
|
class SandboxedFormatter(Formatter):
|
||
|
|
def __init__(self, env: Environment, **kwargs: t.Any) -> None:
|
||
|
|
self._env = env
|
||
|
|
super().__init__(**kwargs)
|
||
|
|
|
||
|
|
def get_field(
|
||
|
|
self, field_name: str, args: t.Sequence[t.Any], kwargs: t.Mapping[str, t.Any]
|
||
|
|
) -> t.Tuple[t.Any, str]:
|
||
|
|
first, rest = formatter_field_name_split(field_name)
|
||
|
|
obj = self.get_value(first, args, kwargs)
|
||
|
|
for is_attr, i in rest:
|
||
|
|
if is_attr:
|
||
|
|
obj = self._env.getattr(obj, i)
|
||
|
|
else:
|
||
|
|
obj = self._env.getitem(obj, i)
|
||
|
|
return obj, first
|
||
|
|
|
||
|
|
|
||
|
|
class SandboxedEscapeFormatter(SandboxedFormatter, EscapeFormatter):
|
||
|
|
pass
|