/usr/share/pyshared/circuits/web/tools.py is in python-circuits 2.1.0-2.
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 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 | # Module: tools
# Date: 16th February 2009
# Author: James Mills, prologic at shortcircuit dot net dot au
"""Tools
This module implements tools used throughout circuits.web.
These tools can also be used within Controlelrs and request handlers.
"""
import os
import stat
import hashlib
import mimetypes
from time import mktime
from email.utils import formatdate
from datetime import datetime, timedelta
from email.generator import _make_boundary
import collections
mimetypes.init()
mimetypes.add_type("image/x-dwg", ".dwg")
mimetypes.add_type("image/x-icon", ".ico")
mimetypes.add_type("application/xhtml+xml", ".xhtml")
from . import _httpauth
from .utils import get_ranges, compress
from .errors import HTTPError, NotFound, Redirect, Unauthorized
def expires(request, response, secs=0, force=False):
"""Tool for influencing cache mechanisms using the 'Expires' header.
'secs' must be either an int or a datetime.timedelta, and indicates the
number of seconds between response.time and when the response should
expire. The 'Expires' header will be set to (response.time + secs).
If 'secs' is zero, the 'Expires' header is set one year in the past, and
the following "cache prevention" headers are also set:
- 'Pragma': 'no-cache'
- 'Cache-Control': 'no-cache, must-revalidate'
If 'force' is False (the default), the following headers are checked:
'Etag', 'Last-Modified', 'Age', 'Expires'. If any are already present,
none of the above response headers are set.
"""
headers = response.headers
cacheable = False
if not force:
# some header names that indicate that the response can be cached
for indicator in ('Etag', 'Last-Modified', 'Age', 'Expires'):
if indicator in headers:
cacheable = True
break
if not cacheable:
if isinstance(secs, timedelta):
secs = (86400 * secs.days) + secs.seconds
if secs == 0:
if force or "Pragma" not in headers:
headers["Pragma"] = "no-cache"
if request.protocol >= (1, 1):
if force or "Cache-Control" not in headers:
headers["Cache-Control"] = "no-cache, must-revalidate"
# Set an explicit Expires date in the past.
now = datetime.now()
lastyear = now.replace(year=now.year - 1)
expiry = formatdate(
mktime(lastyear.timetuple()), usegmt=True
)
else:
expiry = formatdate(response.time + secs, usegmt=True)
if force or "Expires" not in headers:
headers["Expires"] = expiry
def serve_file(request, response, path, type=None, disposition=None,
name=None):
"""Set status, headers, and body in order to serve the given file.
The Content-Type header will be set to the type arg, if provided.
If not provided, the Content-Type will be guessed by the file extension
of the 'path' argument.
If disposition is not None, the Content-Disposition header will be set
to "<disposition>; filename=<name>". If name is None, it will be set
to the basename of path. If disposition is None, no Content-Disposition
header will be written.
"""
if not os.path.isabs(path):
raise ValueError("'%s' is not an absolute path." % path)
try:
st = os.stat(path)
except OSError:
return NotFound(request, response)
# Check if path is a directory.
if stat.S_ISDIR(st.st_mode):
# Let the caller deal with it as they like.
return NotFound(request, response)
# Set the Last-Modified response header, so that
# modified-since validation code can work.
response.headers['Last-Modified'] = formatdate(
st.st_mtime, usegmt=True
)
result = validate_since(request, response)
if result is not None:
return result
if type is None:
# Set content-type based on filename extension
ext = ""
i = path.rfind('.')
if i != -1:
ext = path[i:].lower()
type = mimetypes.types_map.get(ext, "text/plain")
response.headers['Content-Type'] = type
if disposition is not None:
if name is None:
name = os.path.basename(path)
cd = '%s; filename="%s"' % (disposition, name)
response.headers["Content-Disposition"] = cd
# Set Content-Length and use an iterable (file object)
# this way CP won't load the whole file in memory
c_len = st.st_size
bodyfile = open(path, 'rb')
# HTTP/1.0 didn't have Range/Accept-Ranges headers, or the 206 code
if request.protocol >= (1, 1):
response.headers["Accept-Ranges"] = "bytes"
r = get_ranges(request.headers.get('Range'), c_len)
if r == []:
response.headers['Content-Range'] = "bytes */%s" % c_len
return HTTPError(request, response, 416)
if r:
if len(r) == 1:
# Return a single-part response.
start, stop = r[0]
r_len = stop - start
response.code = 206
response.headers['Content-Range'] = (
"bytes %s-%s/%s" % (start, stop - 1, c_len)
)
response.headers['Content-Length'] = r_len
bodyfile.seek(start)
response.body = bodyfile.read(r_len)
else:
# Return a multipart/byteranges response.
response.code = 206
boundary = _make_boundary()
ct = "multipart/byteranges; boundary=%s" % boundary
response.headers['Content-Type'] = ct
if "Content-Length" in response.headers:
# Delete Content-Length header so finalize() recalcs it.
del response.headers["Content-Length"]
def file_ranges():
# Apache compatibility:
yield "\r\n"
for start, stop in r:
yield "--" + boundary
yield "\r\nContent-type: %s" % type
yield ("\r\nContent-range: bytes %s-%s/%s\r\n\r\n"
% (start, stop - 1, c_len))
bodyfile.seek(start)
yield bodyfile.read(stop - start)
yield "\r\n"
# Final boundary
yield "--" + boundary + "--"
# Apache compatibility:
yield "\r\n"
response.body = file_ranges()
else:
response.headers['Content-Length'] = c_len
response.body = bodyfile
else:
response.headers['Content-Length'] = c_len
response.body = bodyfile
return response
def serve_download(request, response, path, name=None):
"""Serve 'path' as an application/x-download attachment."""
type = "application/x-download"
disposition = "attachment"
return serve_file(request, response, path, type, disposition, name)
def validate_etags(request, response, autotags=False):
"""Validate the current ETag against If-Match, If-None-Match headers.
If autotags is True, an ETag response-header value will be provided
from an MD5 hash of the response body (unless some other code has
already provided an ETag header). If False (the default), the ETag
will not be automatic.
WARNING: the autotags feature is not designed for URL's which allow
methods other than GET. For example, if a POST to the same URL returns
no content, the automatic ETag will be incorrect, breaking a fundamental
use for entity tags in a possibly destructive fashion. Likewise, if you
raise 304 Not Modified, the response body will be empty, the ETag hash
will be incorrect, and your application will break.
See http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.24
"""
# Guard against being run twice.
if hasattr(response, "ETag"):
return
code = response.code
etag = response.headers.get('ETag')
# Automatic ETag generation. See warning in docstring.
if (not etag) and autotags:
if code == 200:
etag = response.collapse_body()
etag = '"%s"' % hashlib.md5.new(etag).hexdigest()
response.headers['ETag'] = etag
response.ETag = etag
# "If the request would, without the If-Match header field, result in
# anything other than a 2xx or 412 status, then the If-Match header
# MUST be ignored."
if code >= 200 and code <= 299:
conditions = request.headers.elements('If-Match') or []
conditions = [str(x) for x in conditions]
if conditions and not (conditions == ["*"] or etag in conditions):
return HTTPError(
request, response, 412,
description="If-Match failed: ETag %r did not match %r" % (
etag, conditions
)
)
conditions = request.headers.elements('If-None-Match') or []
conditions = [str(x) for x in conditions]
if conditions == ["*"] or etag in conditions:
if request.method in ("GET", "HEAD"):
return Redirect(request, response, [], code=304)
else:
return HTTPError(
request, response, 412,
description=(
"If-None-Match failed: ETag %r matched %r" % (
etag, conditions
)
)
)
def validate_since(request, response):
"""Validate the current Last-Modified against If-Modified-Since headers.
If no code has set the Last-Modified response header, then no validation
will be performed.
"""
lastmod = response.headers.get('Last-Modified')
if lastmod:
code = response.code
since = request.headers.get('If-Unmodified-Since')
if since and since != lastmod:
if (code >= 200 and code <= 299) or code == 412:
return HTTPError(request, response, 412)
since = request.headers.get('If-Modified-Since')
if since and since == lastmod:
if (code >= 200 and code <= 299) or code == 304:
if request.method in ("GET", "HEAD"):
return Redirect(request, response, [], code=304)
else:
return HTTPError(request, response, 412)
def check_auth(request, response, realm, users, encrypt=None):
"""Check Authentication
If an authorization header contains credentials, return True, else False.
:param realm: The authentication realm.
:type realm: str
:param users: A dict of the form: {username: password} or a callable
returning a dict.
:type users: dict or callable
:param encrypt: Callable used to encrypt the password returned from
the user-agent. if None it defaults to a md5 encryption.
:type encrypt: callable
"""
if 'authorization' in request.headers:
# make sure the provided credentials are correctly set
ah = _httpauth.parseAuthorization(request.headers['authorization'])
if ah is None:
return HTTPError(request, response, 400)
if not encrypt:
encrypt = _httpauth.DIGEST_AUTH_ENCODERS[_httpauth.MD5]
if isinstance(users, collections.Callable):
try:
# backward compatibility
users = users() # expect it to return a dictionary
if not isinstance(users, dict):
raise ValueError("Authentication users must be a dict")
# fetch the user password
password = users.get(ah["username"], None)
except TypeError:
# returns a password (encrypted or clear text)
password = users(ah["username"])
else:
if not isinstance(users, dict):
raise ValueError("Authentication users must be a dict")
# fetch the user password
password = users.get(ah["username"], None)
# validate the authorization by re-computing it here
# and compare it with what the user-agent provided
if _httpauth.checkResponse(ah, password, method=request.method,
encrypt=encrypt, realm=realm):
request.login = ah["username"]
return True
request.login = False
return False
def basic_auth(request, response, realm, users, encrypt=None):
"""Perform Basic Authentication
If auth fails, returns an Unauthorized error with a
basic authentication header.
:param realm: The authentication realm.
:type realm: str
:param users: A dict of the form: {username: password} or a callable
returning a dict.
:type users: dict or callable
:param encrypt: Callable used to encrypt the password returned from
the user-agent. if None it defaults to a md5 encryption.
:type encrypt: callable
"""
if check_auth(request, response, realm, users, encrypt):
return
# inform the user-agent this path is protected
response.headers["WWW-Authenticate"] = _httpauth.basicAuth(realm)
return Unauthorized(request, response)
def digest_auth(request, response, realm, users):
"""Perform Digest Authentication
If auth fails, raise 401 with a digest authentication header.
:param realm: The authentication realm.
:type realm: str
:param users: A dict of the form: {username: password} or a callable
returning a dict.
:type users: dict or callable
"""
if check_auth(request, response, realm, users):
return
# inform the user-agent this path is protected
response.headers["WWW-Authenticate"] = _httpauth.digestAuth(realm)
return Unauthorized(request, response)
def gzip(response, level=4, mime_types=['text/html', 'text/plain']):
"""Try to gzip the response body if Content-Type in mime_types.
response.headers['Content-Type'] must be set to one of the
values in the mime_types arg before calling this function.
No compression is performed if any of the following hold:
* The client sends no Accept-Encoding request header
* No 'gzip' or 'x-gzip' is present in the Accept-Encoding header
* No 'gzip' or 'x-gzip' with a qvalue > 0 is present
* The 'identity' value is given with a qvalue > 0.
"""
if not response.body:
# Response body is empty (might be a 304 for instance)
return response
# If returning cached content (which should already have been gzipped),
# don't re-zip.
if getattr(response.request, "cached", False):
return response
acceptable = response.request.headers.elements('Accept-Encoding')
if not acceptable:
# If no Accept-Encoding field is present in a request,
# the server MAY assume that the client will accept any
# content coding. In this case, if "identity" is one of
# the available content-codings, then the server SHOULD use
# the "identity" content-coding, unless it has additional
# information that a different content-coding is meaningful
# to the client.
return response
ct = response.headers.get('Content-Type', 'text/html').split(';')[0]
for coding in acceptable:
if coding.value == 'identity' and coding.qvalue != 0:
return response
if coding.value in ('gzip', 'x-gzip'):
if coding.qvalue == 0:
return response
if ct in mime_types:
# Return a generator that compresses the page
varies = response.headers.get("Vary", "")
varies = [x.strip() for x in varies.split(",") if x.strip()]
if "Accept-Encoding" not in varies:
varies.append("Accept-Encoding")
response.headers['Vary'] = ", ".join(varies)
response.headers['Content-Encoding'] = 'gzip'
response.body = compress(response.body, level)
if "Content-Length" in response.headers:
# Delete Content-Length header so finalize() recalcs it.
del response.headers["Content-Length"]
return response
return HTTPError(
response.request, response, 406, description="identity, gzip"
)
|