github.com
/
pypiserver
mirror of https://github.com/pypiserver/pypiserver
1
0
Fork 1
Code Releases Wiki Activity

FMT: pypiserver/_app.py

This commit is contained in:
Matthew Planchard 2019-09-17 20:43:43 -05:00
parent a52c0d6f4c
commit 957538a260
No known key found for this signature in database
GPG Key ID: 4ABC11DF33394F00
1 changed files with 132 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

191
pypiserver/_app.py
View File

@ -8,7 +8,15 @@ import xml.dom.minidom
from . import __version__
from . import core
from .bottle import static_file, redirect, request, response, HTTPError, Bottle, template
from .bottle import (
static_file,
redirect,
request,
response,
HTTPError,
Bottle,
template,
)
try:
import xmlrpc.client as xmlrpclib # py3
@ -40,7 +48,6 @@ class auth(object):
self.action = action
def __call__(self, method):
def protector(*args, **kwargs):
if self.action in config.authenticated:
if not request.auth or request.auth[1] is None:
@ -54,25 +61,32 @@ class auth(object):
return protector
@app.hook('before_request')
@app.hook("before_request")
def log_request():
log.info(config.log_req_frmt, request.environ)
@app.hook('before_request')
@app.hook("before_request")
def print_request():
parsed = urlparse(request.urlparts.scheme + "://" + request.urlparts.netloc)
request.custom_host = parsed.netloc
request.custom_fullpath = parsed.path.rstrip('/') + '/' + request.fullpath.lstrip('/')
request.custom_fullpath = (
parsed.path.rstrip("/") + "/" + request.fullpath.lstrip("/")
)
@app.hook('after_request')
@app.hook("after_request")
def log_response():
log.info(config.log_res_frmt, { # vars(response)) ## DOES NOT WORK!
'response': response,
'status': response.status, 'headers': response.headers,
'body': response.body, 'cookies': response._cookies,
})
log.info(
config.log_res_frmt,
{ # vars(response)) ## DOES NOT WORK!
"response": response,
"status": response.status,
"headers": response.headers,
"body": response.body,
"cookies": response._cookies,
},
)
@app.error
@ -85,7 +99,7 @@ def favicon():
return HTTPError(404)
@app.route('/')
@app.route("/")
def root():
fp = request.custom_fullpath
@ -95,16 +109,18 @@ def root():
numpkgs = 0
# Ensure template() does not consider `msg` as filename!
msg = config.welcome_msg + '\n'
return template(msg,
URL=request.url.rstrip("/") + '/',
msg = config.welcome_msg + "\n"
return template(
msg,
URL=request.url.rstrip("/") + "/",
VERSION=__version__,
NUMPKGS=numpkgs,
PACKAGES=fp.rstrip("/") + "/packages/",
SIMPLE=fp.rstrip("/") + "/simple/"
SIMPLE=fp.rstrip("/") + "/simple/",
)
_bottle_upload_filename_re = re.compile(r'^[a-z0-9_.!+-]+$', re.I)
_bottle_upload_filename_re = re.compile(r"^[a-z0-9_.!+-]+$", re.I)
def is_valid_pkg_filename(fname):
@ -114,13 +130,13 @@ def is_valid_pkg_filename(fname):
def doc_upload():
try:
content = request.files['content']
content = request.files["content"]
except KeyError:
raise HTTPError(400, "Missing 'content' file-field!")
zip_data = content.file.read()
try:
zf = zipfile.ZipFile(BytesIO(zip_data))
zf.getinfo('index.html')
zf.getinfo("index.html")
except Exception:
raise HTTPError(400, "not a zip file")
@ -141,47 +157,61 @@ def remove_pkg():
os.unlink(found.fn)
Upload = namedtuple('Upload', 'pkg sig')
Upload = namedtuple("Upload", "pkg sig")
def file_upload():
ufiles = Upload._make(
request.files.get(f, None) for f in ('content', 'gpg_signature'))
request.files.get(f, None) for f in ("content", "gpg_signature")
)
if not ufiles.pkg:
raise HTTPError(400, "Missing 'content' file-field!")
if (ufiles.sig and
'%s.asc' % ufiles.pkg.raw_filename != ufiles.sig.raw_filename):
raise HTTPError(400, "Unrelated signature %r for package %r!",
ufiles.sig, ufiles.pkg)
if (
ufiles.sig
and "%s.asc" % ufiles.pkg.raw_filename != ufiles.sig.raw_filename
):
raise HTTPError(
400,
"Unrelated signature %r for package %r!",
ufiles.sig,
ufiles.pkg,
)
for uf in ufiles:
if not uf:
continue
if (not is_valid_pkg_filename(uf.raw_filename) or
core.guess_pkgname_and_version(uf.raw_filename) is None):
if (
not is_valid_pkg_filename(uf.raw_filename)
or core.guess_pkgname_and_version(uf.raw_filename) is None
):
raise HTTPError(400, "Bad filename: %s" % uf.raw_filename)
if not config.overwrite and core.exists(packages.root, uf.raw_filename):
log.warn("Cannot upload %r since it already exists! \n"
log.warn(
"Cannot upload %r since it already exists! \n"
" You may start server with `--overwrite` option. ",
uf.raw_filename)
raise HTTPError(409, "Package %r already exists!\n"
uf.raw_filename,
)
raise HTTPError(
409,
"Package %r already exists!\n"
" You may start server with `--overwrite` option.",
uf.raw_filename)
uf.raw_filename,
)
core.store(packages.root, uf.raw_filename, uf.save)
if request.auth:
user = request.auth[0]
else:
user = 'anon'
log.info('User %r stored %r.', user, uf.raw_filename)
user = "anon"
log.info("User %r stored %r.", user, uf.raw_filename)
@app.post('/')
@app.post("/")
@auth("update")
def update():
try:
action = request.forms[':action']
action = request.forms[":action"]
except KeyError:
raise HTTPError(400, "Missing ':action' field!")
@ -201,35 +231,46 @@ def update():
@app.route("/simple")
@app.route("/simple/:prefix")
@app.route('/packages')
@app.route("/packages")
@auth("list")
def pep_503_redirects(prefix=None):
return redirect(request.custom_fullpath + "/", 301)
@app.post('/RPC2')
@app.post("/RPC2")
@auth("list")
def handle_rpc():
"""Handle pip-style RPC2 search requests"""
parser = xml.dom.minidom.parse(request.body)
methodname = parser.getElementsByTagName(
"methodName")[0].childNodes[0].wholeText.strip()
methodname = (
parser.getElementsByTagName("methodName")[0]
.childNodes[0]
.wholeText.strip()
)
log.info("Processing RPC2 request for '%s'", methodname)
if methodname == 'search':
value = parser.getElementsByTagName(
"string")[0].childNodes[0].wholeText.strip()
if methodname == "search":
value = (
parser.getElementsByTagName("string")[0]
.childNodes[0]
.wholeText.strip()
)
response = []
ordering = 0
for p in packages():
if p.pkgname.count(value) > 0:
# We do not presently have any description/summary, returning
# version instead
d = {'_pypi_ordering': ordering, 'version': p.version,
'name': p.pkgname, 'summary': p.version}
d = {
"_pypi_ordering": ordering,
"version": p.version,
"name": p.pkgname,
"summary": p.version,
}
response.append(d)
ordering += 1
call_string = xmlrpclib.dumps((response,), 'search',
methodresponse=True)
call_string = xmlrpclib.dumps(
(response,), "search", methodresponse=True
)
return call_string
@ -259,19 +300,29 @@ def simple(prefix=""):
# PEP 503: require normalized prefix
normalized = core.normalize_pkgname_for_url(prefix)
if prefix != normalized:
return redirect('/simple/{0}/'.format(normalized), 301)
return redirect("/simple/{0}/".format(normalized), 301)
files = sorted(core.find_packages(packages(), prefix=prefix),
key=lambda x: (x.parsed_version, x.relfn))
files = sorted(
core.find_packages(packages(), prefix=prefix),
key=lambda x: (x.parsed_version, x.relfn),
)
if not files:
if config.redirect_to_fallback:
return redirect("%s/%s/" % (config.fallback_url.rstrip("/"), prefix))
return HTTPError(404, 'Not Found (%s does not exist)\n\n' % normalized)
return redirect(
"%s/%s/" % (config.fallback_url.rstrip("/"), prefix)
)
return HTTPError(404, "Not Found (%s does not exist)\n\n" % normalized)
fp = request.custom_fullpath
links = [(os.path.basename(f.relfn),
urljoin(fp, "../../packages/%s" % f.fname_and_hash(config.hash_algo)))
for f in files]
links = [
(
os.path.basename(f.relfn),
urljoin(
fp, "../../packages/%s" % f.fname_and_hash(config.hash_algo)
),
)
for f in files
]
tmpl = """\
<html>
<head>
@ -288,16 +339,18 @@ def simple(prefix=""):
return template(tmpl, prefix=prefix, links=links)
@app.route('/packages/')
@app.route("/packages/")
@auth("list")
def list_packages():
fp = request.custom_fullpath
files = sorted(core.find_packages(packages()),
key=lambda x: (os.path.dirname(x.relfn),
x.pkgname,
x.parsed_version))
links = [(f.relfn_unix, urljoin(fp, f.fname_and_hash(config.hash_algo)))
for f in files]
files = sorted(
core.find_packages(packages()),
key=lambda x: (os.path.dirname(x.relfn), x.pkgname, x.parsed_version),
)
links = [
(f.relfn_unix, urljoin(fp, f.fname_and_hash(config.hash_algo)))
for f in files
]
tmpl = """\
<html>
<head>
@ -314,7 +367,7 @@ def list_packages():
return template(tmpl, links=links)
@app.route('/packages/:filename#.*#')
@app.route("/packages/:filename#.*#")
@auth("download")
def server_static(filename):
entries = core.find_packages(packages())
@ -322,17 +375,21 @@ def server_static(filename):
f = x.relfn_unix
if f == filename:
response = static_file(
filename, root=x.root, mimetype=mimetypes.guess_type(filename)[0])
filename,
root=x.root,
mimetype=mimetypes.guess_type(filename)[0],
)
if config.cache_control:
response.set_header(
"Cache-Control", "public, max-age=%s" % config.cache_control)
"Cache-Control", "public, max-age=%s" % config.cache_control
)
return response
return HTTPError(404, 'Not Found (%s does not exist)\n\n' % filename)
return HTTPError(404, "Not Found (%s does not exist)\n\n" % filename)
@app.route('/:prefix')
@app.route('/:prefix/')
@app.route("/:prefix")
@app.route("/:prefix/")
def bad_url(prefix):
"""Redirect unknown root URLs to /simple/."""
return redirect(core.get_bad_url_redirect_path(request, prefix))