forked from github.com/pypiserver
1375a67c55
Resolves #237 Previously, we were not running any sort of URL escaping on values passed in from the client that were used for redirects. This allowed injection attacks via URL encoded newlines in the original request. This update ensures that all user-supplied paths that are used as components of redirects are passed through `urllib.parse.quote()` (or the python 2 equivalent) prior to being used in a redirect response. Also specified 127.0.0.1 rather than 0.0.0.0 (the default) in server tests to avoid triggering firewall dialogs when testing on MacOS
432 lines
12 KiB
Python
432 lines
12 KiB
Python
#! /usr/bin/env py.test
|
|
"""
|
|
Checks an actual pypi-server against various clients.
|
|
|
|
The tests below are using 3 ways to startup pypi-servers:
|
|
|
|
- "open": a per-module server instance without any authed operations,
|
|
serving a single `wheel` package, on a fixed port.
|
|
- "open": a per-module server instance with authed 'download/upload'
|
|
operations, serving a single `wheel` package, on a fixed port.
|
|
- "new_server": starting a new server with any configurations on each test.
|
|
|
|
"""
|
|
from __future__ import print_function
|
|
|
|
from collections import namedtuple
|
|
import contextlib
|
|
import functools
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from shlex import split
|
|
from subprocess import Popen
|
|
from textwrap import dedent
|
|
try:
|
|
from urllib.request import urlopen
|
|
except ImportError:
|
|
from urllib import urlopen
|
|
|
|
from py import path # @UnresolvedImport
|
|
import pytest
|
|
|
|
|
|
# ######################################################################
|
|
# Fixtures & Helper Functions
|
|
# ######################################################################
|
|
|
|
|
|
_BUFF_SIZE = 2**16
|
|
_port = 8090
|
|
SLEEP_AFTER_SRV = 3 # sec
|
|
|
|
|
|
@pytest.fixture
|
|
def port():
|
|
global _port
|
|
_port += 1
|
|
return _port
|
|
|
|
|
|
Srv = namedtuple('Srv', ('proc', 'port', 'package'))
|
|
|
|
|
|
def _run_server(packdir, port, authed, other_cli=''):
|
|
"""Run a server, optionally with partial auth enabled."""
|
|
pswd_opt_choices = {
|
|
True: "-Ptests/htpasswd.a.a -a update,download",
|
|
False: "-P. -a.",
|
|
'partial': "-Ptests/htpasswd.a.a -a update",
|
|
}
|
|
pswd_opts = pswd_opt_choices[authed]
|
|
cmd = (
|
|
"%s -m pypiserver.__main__ -vvv --overwrite -i 127.0.0.1 "
|
|
"-p %s %s %s %s" % (
|
|
sys.executable,
|
|
port,
|
|
pswd_opts,
|
|
other_cli,
|
|
packdir,
|
|
)
|
|
)
|
|
proc = subprocess.Popen(cmd.split(), bufsize=_BUFF_SIZE)
|
|
time.sleep(SLEEP_AFTER_SRV)
|
|
assert proc.poll() is None
|
|
|
|
return Srv(proc, int(port), packdir)
|
|
|
|
|
|
def _kill_server(srv):
|
|
print('Killing %s' % (srv,))
|
|
try:
|
|
srv.proc.terminate()
|
|
time.sleep(1)
|
|
finally:
|
|
srv.proc.kill()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def new_server(packdir, port, authed=False, other_cli=''):
|
|
srv = _run_server(packdir, port,
|
|
authed=authed, other_cli=other_cli)
|
|
try:
|
|
yield srv
|
|
finally:
|
|
_kill_server(srv)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def chdir(d):
|
|
old_d = os.getcwd()
|
|
try:
|
|
os.chdir(d)
|
|
yield
|
|
finally:
|
|
os.chdir(old_d)
|
|
|
|
|
|
def _run_python(cmd):
|
|
ncmd = '%s %s' % (sys.executable, cmd)
|
|
return os.system(ncmd)
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def project(request):
|
|
def fin():
|
|
tmpdir.remove(True)
|
|
tmpdir = path.local(tempfile.mkdtemp())
|
|
request.addfinalizer(fin)
|
|
src_setup_py = path.local().join('tests', 'centodeps-setup.py')
|
|
assert src_setup_py.check()
|
|
projdir = tmpdir.join('centodeps')
|
|
projdir.mkdir()
|
|
dst_setup_py = projdir.join('setup.py')
|
|
src_setup_py.copy(dst_setup_py)
|
|
assert dst_setup_py.check()
|
|
|
|
return projdir
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def package(project, request):
|
|
with chdir(project.strpath):
|
|
cmd = 'setup.py bdist_wheel'
|
|
assert _run_python(cmd) == 0
|
|
pkgs = list(project.join('dist').visit('centodeps*.whl'))
|
|
assert len(pkgs) == 1
|
|
pkg = path.local(pkgs[0])
|
|
assert pkg.check()
|
|
|
|
return pkg
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def packdir(package):
|
|
return package.dirpath()
|
|
|
|
|
|
open_port = 8081
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def open_server(packdir, request):
|
|
srv = _run_server(packdir, open_port, authed=False)
|
|
fin = functools.partial(_kill_server, srv)
|
|
request.addfinalizer(fin)
|
|
|
|
return srv
|
|
|
|
|
|
protected_port = 8082
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def protected_server(packdir, request):
|
|
srv = _run_server(packdir, protected_port, authed=True)
|
|
fin = functools.partial(_kill_server, srv)
|
|
request.addfinalizer(fin)
|
|
|
|
return srv
|
|
|
|
|
|
@pytest.fixture
|
|
def empty_packdir(tmpdir):
|
|
return tmpdir.mkdir("dists")
|
|
|
|
|
|
def _build_url(port, user='', pswd=''):
|
|
auth = '%s:%s@' % (user, pswd) if user or pswd else ''
|
|
return 'http://%slocalhost:%s' % (auth, port)
|
|
|
|
|
|
def _run_pip(cmd):
|
|
ncmd = (
|
|
"pip --disable-pip-version-check --retries 0 --timeout 5 --no-input %s"
|
|
) % cmd
|
|
print('PIP: %s' % ncmd)
|
|
proc = Popen(split(ncmd))
|
|
proc.communicate()
|
|
return proc.returncode
|
|
|
|
|
|
def _run_pip_install(cmd, port, install_dir, user=None, pswd=None):
|
|
url = _build_url(port, user, pswd)
|
|
# ncmd = '-vv install --download %s -i %s %s' % (install_dir, url, cmd)
|
|
ncmd = '-vv download -d %s -i %s %s' % (install_dir, url, cmd)
|
|
return _run_pip(ncmd)
|
|
|
|
|
|
@pytest.fixture
|
|
def pipdir(tmpdir):
|
|
return tmpdir.mkdir("pip")
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def pypirc_tmpfile(port, user, password):
|
|
"""Create a temporary pypirc file."""
|
|
fd, filepath = tempfile.mkstemp()
|
|
os.close(fd)
|
|
with open(filepath, 'w') as rcfile:
|
|
rcfile.writelines(
|
|
'\n'.join((
|
|
'[distutils]',
|
|
'index-servers: test',
|
|
''
|
|
'[test]',
|
|
'repository: {}'.format(_build_url(port)),
|
|
'username: {}'.format(user),
|
|
'password: {}'.format(password),
|
|
))
|
|
)
|
|
with open(filepath) as rcfile:
|
|
print(rcfile.read())
|
|
yield filepath
|
|
os.remove(filepath)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def pypirc_file(txt):
|
|
pypirc_path = path.local('~/.pypirc', expanduser=1)
|
|
old_pypirc = pypirc_path.read() if pypirc_path.check() else None
|
|
pypirc_path.write(txt)
|
|
try:
|
|
yield
|
|
finally:
|
|
if old_pypirc:
|
|
pypirc_path.write(old_pypirc)
|
|
else:
|
|
pypirc_path.remove()
|
|
|
|
|
|
def twine_upload(packages, repository='test', conf='pypirc',
|
|
expect_failure=False):
|
|
"""Call 'twine upload' with appropriate arguments"""
|
|
proc = Popen((
|
|
'twine',
|
|
'upload',
|
|
'--repository', repository,
|
|
'--config-file', conf,
|
|
' '.join(packages),
|
|
))
|
|
proc.communicate()
|
|
if not expect_failure and proc.returncode:
|
|
assert False, 'Twine upload failed. See stdout/err'
|
|
|
|
|
|
def twine_register(packages, repository='test', conf='pypirc',
|
|
expect_failure=False):
|
|
"""Call 'twine register' with appropriate args"""
|
|
proc = Popen((
|
|
'twine',
|
|
'register',
|
|
'--repository', repository,
|
|
'--config-file', conf,
|
|
' '.join(packages)
|
|
))
|
|
proc.communicate()
|
|
if not expect_failure and proc.returncode:
|
|
assert False, 'Twine register failed. See stdout/err'
|
|
|
|
|
|
# ######################################################################
|
|
# Tests
|
|
# ######################################################################
|
|
|
|
|
|
def test_pipInstall_packageNotFound(empty_packdir, port, pipdir, package):
|
|
with new_server(empty_packdir, port):
|
|
cmd = "centodeps"
|
|
assert _run_pip_install(cmd, port, pipdir) != 0
|
|
assert not pipdir.listdir()
|
|
|
|
|
|
def test_pipInstall_openOk(open_server, package, pipdir):
|
|
cmd = "centodeps"
|
|
assert _run_pip_install(cmd, open_server.port, pipdir) == 0
|
|
assert pipdir.join(package.basename).check()
|
|
|
|
|
|
def test_pipInstall_authedFails(protected_server, pipdir):
|
|
cmd = "centodeps"
|
|
assert _run_pip_install(cmd, protected_server.port, pipdir) != 0
|
|
assert not pipdir.listdir()
|
|
|
|
|
|
def test_pipInstall_authedOk(protected_server, package, pipdir):
|
|
cmd = "centodeps"
|
|
assert _run_pip_install(cmd, protected_server.port, pipdir,
|
|
user='a', pswd='a') == 0
|
|
assert pipdir.join(package.basename).check()
|
|
|
|
|
|
@pytest.mark.parametrize("pkg_frmt", ['bdist', 'bdist_wheel'])
|
|
def test_setuptoolsUpload_open(empty_packdir, port, project, package,
|
|
pkg_frmt):
|
|
url = _build_url(port, None, None)
|
|
with pypirc_file(dedent("""\
|
|
[distutils]
|
|
index-servers: test
|
|
|
|
[test]
|
|
repository: %s
|
|
username: ''
|
|
password: ''
|
|
""" % url)):
|
|
with new_server(empty_packdir, port):
|
|
with chdir(project.strpath):
|
|
cmd = "setup.py -vvv %s upload -r %s" % (pkg_frmt, url)
|
|
for i in range(5):
|
|
print('++Attempt #%s' % i)
|
|
assert _run_python(cmd) == 0
|
|
time.sleep(SLEEP_AFTER_SRV)
|
|
assert len(empty_packdir.listdir()) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("pkg_frmt", ['bdist', 'bdist_wheel'])
|
|
def test_setuptoolsUpload_authed(empty_packdir, port, project, package,
|
|
pkg_frmt, monkeypatch):
|
|
url = _build_url(port)
|
|
with pypirc_file(dedent("""\
|
|
[distutils]
|
|
index-servers: test
|
|
|
|
[test]
|
|
repository: %s
|
|
username: a
|
|
password: a
|
|
""" % url)):
|
|
with new_server(empty_packdir, port, authed=True):
|
|
with chdir(project.strpath):
|
|
cmd = (
|
|
"setup.py -vvv %s register -r "
|
|
"test upload -r test" % pkg_frmt
|
|
)
|
|
for i in range(5):
|
|
print('++Attempt #%s' % i)
|
|
assert _run_python(cmd) == 0
|
|
time.sleep(SLEEP_AFTER_SRV)
|
|
assert len(empty_packdir.listdir()) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("pkg_frmt", ['bdist', 'bdist_wheel'])
|
|
def test_setuptools_upload_partial_authed(empty_packdir, port, project,
|
|
pkg_frmt):
|
|
"""Test uploading a package with setuptools with partial auth."""
|
|
url = _build_url(port)
|
|
with pypirc_file(dedent("""\
|
|
[distutils]
|
|
index-servers: test
|
|
|
|
[test]
|
|
repository: %s
|
|
username: a
|
|
password: a
|
|
""" % url)):
|
|
with new_server(empty_packdir, port, authed='partial'):
|
|
with chdir(project.strpath):
|
|
cmd = ("setup.py -vvv %s register -r test upload -r test" %
|
|
pkg_frmt)
|
|
for i in range(5):
|
|
print('++Attempt #%s' % i)
|
|
assert _run_python(cmd) == 0
|
|
time.sleep(SLEEP_AFTER_SRV)
|
|
assert len(empty_packdir.listdir()) == 1
|
|
|
|
|
|
def test_partial_authed_open_download(empty_packdir, port):
|
|
"""Validate that partial auth still allows downloads."""
|
|
url = _build_url(port) + '/simple'
|
|
with new_server(empty_packdir, port, authed='partial'):
|
|
resp = urlopen(url)
|
|
assert resp.getcode() == 200
|
|
|
|
|
|
def test_twine_upload_open(empty_packdir, port, package):
|
|
"""Test twine upload with no authentication"""
|
|
user, pswd = 'foo', 'bar'
|
|
with new_server(empty_packdir, port):
|
|
with pypirc_tmpfile(port, user, pswd) as rcfile:
|
|
twine_upload([package.strpath], repository='test', conf=rcfile)
|
|
time.sleep(SLEEP_AFTER_SRV)
|
|
assert len(empty_packdir.listdir()) == 1
|
|
|
|
|
|
def test_twine_upload_authed(empty_packdir, port, package):
|
|
"""Test authenticated twine upload"""
|
|
user, pswd = 'a', 'a'
|
|
with new_server(empty_packdir, port, authed=False):
|
|
with pypirc_tmpfile(port, user, pswd) as rcfile:
|
|
twine_upload([package.strpath], repository='test', conf=rcfile)
|
|
time.sleep(SLEEP_AFTER_SRV)
|
|
assert len(empty_packdir.listdir()) == 1
|
|
|
|
assert empty_packdir.join(
|
|
package.basename).check(), (package.basename, empty_packdir.listdir())
|
|
|
|
|
|
def test_twine_upload_partial_authed(empty_packdir, port, package):
|
|
"""Test partially authenticated twine upload"""
|
|
user, pswd = 'a', 'a'
|
|
with new_server(empty_packdir, port, authed='partial'):
|
|
with pypirc_tmpfile(port, user, pswd) as rcfile:
|
|
twine_upload([package.strpath], repository='test', conf=rcfile)
|
|
time.sleep(SLEEP_AFTER_SRV)
|
|
assert len(empty_packdir.listdir()) == 1
|
|
|
|
|
|
def test_twine_register_open(open_server, package):
|
|
"""Test unauthenticated twine registration"""
|
|
srv = open_server
|
|
with pypirc_tmpfile(srv.port, 'foo', 'bar') as rcfile:
|
|
twine_register([package.strpath], repository='test', conf=rcfile)
|
|
|
|
|
|
def test_twine_register_authed_ok(protected_server, package):
|
|
"""Test authenticated twine registration"""
|
|
srv = protected_server
|
|
user, pswd = 'a', 'a'
|
|
with pypirc_tmpfile(srv.port, user, pswd) as rcfile:
|
|
twine_register([package.strpath], repository='test', conf=rcfile)
|