/usr/lib/python2.7/dist-packages/notebook/services/contents/fileio.py is in python-notebook 5.2.2-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 | """
Utilities for file-based Contents/Checkpoints managers.
"""
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
from contextlib import contextmanager
import errno
import io
import os
import shutil
from tornado.web import HTTPError
from notebook.utils import (
to_api_path,
to_os_path,
)
import nbformat
from ipython_genutils.py3compat import str_to_unicode
from traitlets.config import Configurable
from traitlets import Bool
try: #PY3
from base64 import encodebytes, decodebytes
except ImportError: #PY2
from base64 import encodestring as encodebytes, decodestring as decodebytes
def replace_file(src, dst):
""" replace dst with src
switches between os.replace or os.rename based on python 2.7 or python 3
"""
if hasattr(os, 'replace'): # PY3
os.replace(src, dst)
else:
if os.name == 'nt' and os.path.exists(dst):
# Rename over existing file doesn't work on Windows
os.remove(dst)
os.rename(src, dst)
def copy2_safe(src, dst, log=None):
"""copy src to dst
like shutil.copy2, but log errors in copystat instead of raising
"""
shutil.copyfile(src, dst)
try:
shutil.copystat(src, dst)
except OSError:
if log:
log.debug("copystat on %s failed", dst, exc_info=True)
def path_to_intermediate(path):
'''Name of the intermediate file used in atomic writes.
The .~ prefix will make Dropbox ignore the temporary file.'''
dirname, basename = os.path.split(path)
return os.path.join(dirname, '.~'+basename)
def path_to_invalid(path):
'''Name of invalid file after a failed atomic write and subsequent read.'''
dirname, basename = os.path.split(path)
return os.path.join(dirname, basename+'.invalid')
@contextmanager
def atomic_writing(path, text=True, encoding='utf-8', log=None, **kwargs):
"""Context manager to write to a file only if the entire write is successful.
This works by copying the previous file contents to a temporary file in the
same directory, and renaming that file back to the target if the context
exits with an error. If the context is successful, the new data is synced to
disk and the temporary file is removed.
Parameters
----------
path : str
The target file to write to.
text : bool, optional
Whether to open the file in text mode (i.e. to write unicode). Default is
True.
encoding : str, optional
The encoding to use for files opened in text mode. Default is UTF-8.
**kwargs
Passed to :func:`io.open`.
"""
# realpath doesn't work on Windows: http://bugs.python.org/issue9949
# Luckily, we only need to resolve the file itself being a symlink, not
# any of its directories, so this will suffice:
if os.path.islink(path):
path = os.path.join(os.path.dirname(path), os.readlink(path))
tmp_path = path_to_intermediate(path)
if os.path.isfile(path):
copy2_safe(path, tmp_path, log=log)
if text:
# Make sure that text files have Unix linefeeds by default
kwargs.setdefault('newline', '\n')
fileobj = io.open(path, 'w', encoding=encoding, **kwargs)
else:
fileobj = io.open(path, 'wb', **kwargs)
try:
yield fileobj
except:
# Failed! Move the backup file back to the real path to avoid corruption
fileobj.close()
replace_file(tmp_path, path)
raise
# Flush to disk
fileobj.flush()
os.fsync(fileobj.fileno())
fileobj.close()
# Written successfully, now remove the backup copy
if os.path.isfile(tmp_path):
os.remove(tmp_path)
@contextmanager
def _simple_writing(path, text=True, encoding='utf-8', log=None, **kwargs):
"""Context manager to write file without doing atomic writing
( for weird filesystem eg: nfs).
Parameters
----------
path : str
The target file to write to.
text : bool, optional
Whether to open the file in text mode (i.e. to write unicode). Default is
True.
encoding : str, optional
The encoding to use for files opened in text mode. Default is UTF-8.
**kwargs
Passed to :func:`io.open`.
"""
# realpath doesn't work on Windows: http://bugs.python.org/issue9949
# Luckily, we only need to resolve the file itself being a symlink, not
# any of its directories, so this will suffice:
if os.path.islink(path):
path = os.path.join(os.path.dirname(path), os.readlink(path))
if text:
# Make sure that text files have Unix linefeeds by default
kwargs.setdefault('newline', '\n')
fileobj = io.open(path, 'w', encoding=encoding, **kwargs)
else:
fileobj = io.open(path, 'wb', **kwargs)
try:
yield fileobj
except:
fileobj.close()
raise
fileobj.close()
class FileManagerMixin(Configurable):
"""
Mixin for ContentsAPI classes that interact with the filesystem.
Provides facilities for reading, writing, and copying both notebooks and
generic files.
Shared by FileContentsManager and FileCheckpoints.
Note
----
Classes using this mixin must provide the following attributes:
root_dir : unicode
A directory against against which API-style paths are to be resolved.
log : logging.Logger
"""
use_atomic_writing = Bool(True, config=True, help=
"""By default notebooks are saved on disk on a temporary file and then if succefully written, it replaces the old ones.
This procedure, namely 'atomic_writing', causes some bugs on file system whitout operation order enforcement (like some networked fs).
If set to False, the new notebook is written directly on the old one which could fail (eg: full filesystem or quota )""")
@contextmanager
def open(self, os_path, *args, **kwargs):
"""wrapper around io.open that turns permission errors into 403"""
with self.perm_to_403(os_path):
with io.open(os_path, *args, **kwargs) as f:
yield f
@contextmanager
def atomic_writing(self, os_path, *args, **kwargs):
"""wrapper around atomic_writing that turns permission errors to 403.
Depending on flag 'use_atomic_writing', the wrapper perform an actual atomic writing or
simply writes the file (whatever an old exists or not)"""
with self.perm_to_403(os_path):
if self.use_atomic_writing:
with atomic_writing(os_path, *args, log=self.log, **kwargs) as f:
yield f
else:
with _simple_writing(os_path, *args, log=self.log, **kwargs) as f:
yield f
@contextmanager
def perm_to_403(self, os_path=''):
"""context manager for turning permission errors into 403."""
try:
yield
except (OSError, IOError) as e:
if e.errno in {errno.EPERM, errno.EACCES}:
# make 403 error message without root prefix
# this may not work perfectly on unicode paths on Python 2,
# but nobody should be doing that anyway.
if not os_path:
os_path = str_to_unicode(e.filename or 'unknown file')
path = to_api_path(os_path, root=self.root_dir)
raise HTTPError(403, u'Permission denied: %s' % path)
else:
raise
def _copy(self, src, dest):
"""copy src to dest
like shutil.copy2, but log errors in copystat
"""
copy2_safe(src, dest, log=self.log)
def _get_os_path(self, path):
"""Given an API path, return its file system path.
Parameters
----------
path : string
The relative API path to the named file.
Returns
-------
path : string
Native, absolute OS path to for a file.
Raises
------
404: if path is outside root
"""
root = os.path.abspath(self.root_dir)
os_path = to_os_path(path, root)
if not (os.path.abspath(os_path) + os.path.sep).startswith(root):
raise HTTPError(404, "%s is outside root contents directory" % path)
return os_path
def _read_notebook(self, os_path, as_version=4):
"""Read a notebook from an os path."""
with self.open(os_path, 'r', encoding='utf-8') as f:
try:
return nbformat.read(f, as_version=as_version)
except Exception as e:
e_orig = e
# If use_atomic_writing is enabled, we'll guess that it was also
# enabled when this notebook was written and look for a valid
# atomic intermediate.
tmp_path = path_to_intermediate(os_path)
if not self.use_atomic_writing or not os.path.exists(tmp_path):
raise HTTPError(
400,
u"Unreadable Notebook: %s %r" % (os_path, e_orig),
)
# Move the bad file aside, restore the intermediate, and try again.
invalid_file = path_to_invalid(os_path)
replace_file(os_path, invalid_file)
replace_file(tmp_path, os_path)
return self._read_notebook(os_path, as_version)
def _save_notebook(self, os_path, nb):
"""Save a notebook to an os_path."""
with self.atomic_writing(os_path, encoding='utf-8') as f:
nbformat.write(nb, f, version=nbformat.NO_CONVERT)
def _read_file(self, os_path, format):
"""Read a non-notebook file.
os_path: The path to be read.
format:
If 'text', the contents will be decoded as UTF-8.
If 'base64', the raw bytes contents will be encoded as base64.
If not specified, try to decode as UTF-8, and fall back to base64
"""
if not os.path.isfile(os_path):
raise HTTPError(400, "Cannot read non-file %s" % os_path)
with self.open(os_path, 'rb') as f:
bcontent = f.read()
if format is None or format == 'text':
# Try to interpret as unicode if format is unknown or if unicode
# was explicitly requested.
try:
return bcontent.decode('utf8'), 'text'
except UnicodeError:
if format == 'text':
raise HTTPError(
400,
"%s is not UTF-8 encoded" % os_path,
reason='bad format',
)
return encodebytes(bcontent).decode('ascii'), 'base64'
def _save_file(self, os_path, content, format):
"""Save content of a generic file."""
if format not in {'text', 'base64'}:
raise HTTPError(
400,
"Must specify format of file contents as 'text' or 'base64'",
)
try:
if format == 'text':
bcontent = content.encode('utf8')
else:
b64_bytes = content.encode('ascii')
bcontent = decodebytes(b64_bytes)
except Exception as e:
raise HTTPError(
400, u'Encoding error saving %s: %s' % (os_path, e)
)
with self.atomic_writing(os_path, text=False) as f:
f.write(bcontent)
|