upstream/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-06-13 10:40:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge branch 'master' into no-keyauthorization

Browse source
This commit is contained in:
Adrien Ferrand 2019-02-26 22:32:46 +01:00
commit aa400d67f0
30 changed files with 706 additions and 293 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
CHANGELOG.md
View file

@ -14,6 +14,9 @@ Certbot adheres to [Semantic Versioning](https://semver.org/).
warnings described at https://github.com/certbot/josepy/issues/13.
* Apache plugin now respects CERTBOT_DOCS environment variable when adding
command line defaults.
* The running of manual plugin hooks is now always included in Certbot's log
output.
* Tests execution for certbot, certbot-apache and certbot-nginx packages now relies on pytest.
### Fixed
@ -26,6 +29,7 @@ package with changes other than its version number was:
* acme
* certbot
* certbot-apache
* certbot-nginx
More details about these changes can be found on our GitHub repo.

240
acme/examples/http01_example.py Normal file
View file

@ -0,0 +1,240 @@
"""Example ACME-V2 API for HTTP-01 challenge.
Brief:
This a complete usage example of the python-acme API.
Limitations of this example:
- Works for only one Domain name
- Performs only HTTP-01 challenge
- Uses ACME-v2
Workflow:
(Account creation)
- Create account key
- Register account and accept TOS
(Certificate actions)
- Select HTTP-01 within offered challenges by the CA server
- Set up http challenge resource
- Set up standalone web server
- Create domain private key and CSR
- Issue certificate
- Renew certificate
- Revoke certificate
(Account update actions)
- Change contact information
- Deactivate Account
"""
from contextlib import contextmanager
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
import OpenSSL
from acme import challenges
from acme import client
from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone
import josepy as jose
# Constants:
# This is the staging point for ACME-V2 within Let's Encrypt.
DIRECTORY_URL = 'https://acme-staging-v02.api.letsencrypt.org/directory'
USER_AGENT = 'python-acme-example'
# Account key size
ACC_KEY_BITS = 2048
# Certificate private key size
CERT_PKEY_BITS = 2048
# Domain name for the certificate.
DOMAIN = 'client.example.com'
# If you are running Boulder locally, it is possible to configure any port
# number to execute the challenge, but real CA servers will always use port
# 80, as described in the ACME specification.
PORT = 80
# Useful methods and classes:
def new_csr_comp(domain_name, pkey_pem=None):
"""Create certificate signing request."""
if pkey_pem is None:
# Create private key.
pkey = OpenSSL.crypto.PKey()
pkey.generate_key(OpenSSL.crypto.TYPE_RSA, CERT_PKEY_BITS)
pkey_pem = OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM,
pkey)
csr_pem = crypto_util.make_csr(pkey_pem, [domain_name])
return pkey_pem, csr_pem
def select_http01_chall(orderr):
"""Extract authorization resource from within order resource."""
# Authorization Resource: authz.
# This object holds the offered challenges by the server and their status.
authz_list = orderr.authorizations
for authz in authz_list:
# Choosing challenge.
# authz.body.challenges is a set of ChallengeBody objects.
for i in authz.body.challenges:
# Find the supported challenge.
if isinstance(i.chall, challenges.HTTP01):
return i
raise Exception('HTTP-01 challenge was not offered by the CA server.')
@contextmanager
def challenge_server(http_01_resources):
"""Manage standalone server set up and shutdown."""
# Setting up a fake server that binds at PORT and any address.
address = ('', PORT)
try:
servers = standalone.HTTP01DualNetworkedServers(address,
http_01_resources)
# Start client standalone web server.
servers.serve_forever()
yield servers
finally:
# Shutdown client web server and unbind from PORT
servers.shutdown_and_server_close()
def perform_http01(client_acme, challb, orderr):
"""Set up standalone webserver and perform HTTP-01 challenge."""
response, validation = challb.response_and_validation(client_acme.net.key)
resource = standalone.HTTP01RequestHandler.HTTP01Resource(
chall=challb.chall, response=response, validation=validation)
with challenge_server({resource}):
# Let the CA server know that we are ready for the challenge.
client_acme.answer_challenge(challb, response)
# Wait for challenge status and then issue a certificate.
# It is possible to set a deadline time.
finalized_orderr = client_acme.poll_and_finalize(orderr)
return finalized_orderr.fullchain_pem
# Main examples:
def example_http():
"""This example executes the whole process of fulfilling a HTTP-01
challenge for one specific domain.
The workflow consists of:
(Account creation)
- Create account key
- Register account and accept TOS
(Certificate actions)
- Select HTTP-01 within offered challenges by the CA server
- Set up http challenge resource
- Set up standalone web server
- Create domain private key and CSR
- Issue certificate
- Renew certificate
- Revoke certificate
(Account update actions)
- Change contact information
- Deactivate Account
"""
# Create account key
acc_key = jose.JWKRSA(
key=rsa.generate_private_key(public_exponent=65537,
key_size=ACC_KEY_BITS,
backend=default_backend()))
# Register account and accept TOS
net = client.ClientNetwork(acc_key, user_agent=USER_AGENT)
directory = messages.Directory.from_json(net.get(DIRECTORY_URL).json())
client_acme = client.ClientV2(directory, net=net)
# Terms of Service URL is in client_acme.directory.meta.terms_of_service
# Registration Resource: regr
# Creates account with contact information.
email = ('fake@example.com')
regr = client_acme.new_account(
messages.NewRegistration.from_data(
email=email, terms_of_service_agreed=True))
# Create domain private key and CSR
pkey_pem, csr_pem = new_csr_comp(DOMAIN)
# Issue certificate
orderr = client_acme.new_order(csr_pem)
# Select HTTP-01 within offered challenges by the CA server
challb = select_http01_chall(orderr)
# The certificate is ready to be used in the variable "fullchain_pem".
fullchain_pem = perform_http01(client_acme, challb, orderr)
# Renew certificate
_, csr_pem = new_csr_comp(DOMAIN, pkey_pem)
orderr = client_acme.new_order(csr_pem)
challb = select_http01_chall(orderr)
# Performing challenge
fullchain_pem = perform_http01(client_acme, challb, orderr)
# Revoke certificate
fullchain_com = jose.ComparableX509(
OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, fullchain_pem))
try:
client_acme.revoke(fullchain_com, 0) # revocation reason = 0
except errors.ConflictError:
# Certificate already revoked.
pass
# Query registration status.
client_acme.net.account = regr
try:
regr = client_acme.query_registration(regr)
except errors.Error as err:
if err.typ == messages.OLD_ERROR_PREFIX + 'unauthorized' \
or err.typ == messages.ERROR_PREFIX + 'unauthorized':
# Status is deactivated.
pass
raise
# Change contact information
email = 'newfake@example.com'
regr = client_acme.update_registration(
regr.update(
body=regr.body.update(
contact=('mailto:' + email,)
)
)
)
# Deactivate account/registration
regr = client_acme.deactivate_registration(regr)
if __name__ == "__main__":
example_http()

4
acme/setup.py
View file

@ -36,6 +36,7 @@ docs_extras = [
'sphinx_rtd_theme',
]
class PyTest(TestCommand):
user_options = []
@ -50,6 +51,7 @@ class PyTest(TestCommand):
errno = pytest.main(shlex.split(self.pytest_args))
sys.exit(errno)
setup(
name='acme',
version=version,
@ -82,7 +84,7 @@ setup(
'dev': dev_extras,
'docs': docs_extras,
},
tests_require=["pytest"],
test_suite='acme',
tests_require=["pytest"],
cmdclass={"test": PyTest},
)

20
certbot-apache/setup.py
View file

@ -1,5 +1,7 @@
from setuptools import setup
from setuptools import find_packages
from setuptools.command.test import test as TestCommand
import sys
version = '0.32.0.dev0'
@ -21,6 +23,22 @@ docs_extras = [
'sphinx_rtd_theme',
]
class PyTest(TestCommand):
user_options = []
def initialize_options(self):
TestCommand.initialize_options(self)
self.pytest_args = ''
def run_tests(self):
import shlex
# import here, cause outside the eggs aren't loaded
import pytest
errno = pytest.main(shlex.split(self.pytest_args))
sys.exit(errno)
setup(
name='certbot-apache',
version=version,
@ -64,4 +82,6 @@ setup(
],
},
test_suite='certbot_apache',
tests_require=["pytest"],
cmdclass={"test": PyTest},
)

7
certbot-nginx/certbot_nginx/configurator.py
View file

@ -13,6 +13,7 @@ import zope.interface
from acme import challenges
from acme import crypto_util as acme_crypto_util
from certbot import compat
from certbot import constants as core_constants
from certbot import crypto_util
from certbot import errors
@ -164,9 +165,7 @@ class NginxConfigurator(common.Installer):
util.lock_dir_until_exit(self.conf('server-root'))
except (OSError, errors.LockError):
logger.debug('Encountered error:', exc_info=True)
raise errors.PluginError(
'Unable to lock %s', self.conf('server-root'))
raise errors.PluginError('Unable to lock {0}'.format(self.conf('server-root')))
# Entry point in main.py for installing cert
def deploy_cert(self, domain, cert_path, key_path,
@ -899,7 +898,7 @@ class NginxConfigurator(common.Installer):
have permissions of root.
"""
uid = os.geteuid()
uid = compat.os_geteuid()
util.make_or_verify_dir(
self.config.work_dir, core_constants.CONFIG_DIRS_MODE, uid)
util.make_or_verify_dir(

4
certbot-nginx/certbot_nginx/parser.py
View file

@ -81,9 +81,9 @@ class NginxParser(object):
"""
if not os.path.isabs(path):
return os.path.join(self.root, path)
return os.path.normpath(os.path.join(self.root, path))
else:
return path
return os.path.normpath(path)
def _build_addr_to_ssl(self):
"""Builds a map from address to whether it listens on ssl in any server block

19
certbot-nginx/certbot_nginx/tests/configurator_test.py
View file

@ -1,7 +1,6 @@
# pylint: disable=too-many-public-methods
"""Test for certbot_nginx.configurator."""
import os
import shutil
import unittest
import mock
@ -33,12 +32,6 @@ class NginxConfiguratorTest(util.NginxTest):
self.config = util.get_nginx_configurator(
self.config_path, self.config_dir, self.work_dir, self.logs_dir)
def tearDown(self):
shutil.rmtree(self.temp_dir)
shutil.rmtree(self.config_dir)
shutil.rmtree(self.work_dir)
shutil.rmtree(self.logs_dir)
@mock.patch("certbot_nginx.configurator.util.exe_exists")
def test_prepare_no_install(self, mock_exe_exists):
mock_exe_exists.return_value = False
@ -69,8 +62,11 @@ class NginxConfiguratorTest(util.NginxTest):
def test_prepare_locked(self):
server_root = self.config.conf("server-root")
from certbot import util as certbot_util
certbot_util._LOCKS[server_root].release() # pylint: disable=protected-access
self.config.config_test = mock.Mock()
os.remove(os.path.join(server_root, ".certbot.lock"))
certbot_test_util.lock_and_call(self._test_prepare_locked, server_root)
@mock.patch("certbot_nginx.configurator.util.exe_exists")
@ -88,11 +84,11 @@ class NginxConfiguratorTest(util.NginxTest):
def test_get_all_names(self, mock_gethostbyaddr):
mock_gethostbyaddr.return_value = ('155.225.50.69.nephoscale.net', [], [])
names = self.config.get_all_names()
self.assertEqual(names, set(
["155.225.50.69.nephoscale.net", "www.example.org", "another.alias",
self.assertEqual(names, {
"155.225.50.69.nephoscale.net", "www.example.org", "another.alias",
"migration.com", "summer.com", "geese.com", "sslon.com",
"globalssl.com", "globalsslsetssl.com", "ipv6.com", "ipv6ssl.com",
"headers.com"]))
"headers.com"})
def test_supported_enhancements(self):
self.assertEqual(['redirect', 'ensure-http-header', 'staple-ocsp'],
@ -171,6 +167,7 @@ class NginxConfiguratorTest(util.NginxTest):
'abc.www.foo.com': "etc_nginx/foo.conf",
'www.bar.co.uk': "etc_nginx/nginx.conf",
'ipv6.com': "etc_nginx/sites-enabled/ipv6.com"}
conf_path = {key: os.path.normpath(value) for key, value in conf_path.items()}
vhost = self.config.choose_vhosts(name)[0]
path = os.path.relpath(vhost.filep, self.temp_dir)

6
certbot-nginx/certbot_nginx/tests/http_01_test.py
View file

@ -1,6 +1,5 @@
"""Tests for certbot_nginx.http_01"""
import unittest
import shutil
import mock
import six
@ -54,11 +53,6 @@ class HttpPerformTest(util.NginxTest):
from certbot_nginx import http_01
self.http01 = http_01.NginxHttp01(config)
def tearDown(self):
shutil.rmtree(self.temp_dir)
shutil.rmtree(self.config_dir)
shutil.rmtree(self.work_dir)
def test_perform0(self):
responses = self.http01.perform()
self.assertEqual([], responses)

12
certbot-nginx/certbot_nginx/tests/parser_test.py
View file

@ -67,9 +67,15 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
def test_abs_path(self):
nparser = parser.NginxParser(self.config_path)
self.assertEqual('/etc/nginx/*', nparser.abs_path('/etc/nginx/*'))
self.assertEqual(os.path.join(self.config_path, 'foo/bar/'),
nparser.abs_path('foo/bar/'))
if os.name != 'nt':
self.assertEqual('/etc/nginx/*', nparser.abs_path('/etc/nginx/*'))
self.assertEqual(os.path.join(self.config_path, 'foo/bar'),
nparser.abs_path('foo/bar'))
else:
self.assertEqual('C:\\etc\\nginx\\*', nparser.abs_path('C:\\etc\\nginx\\*'))
self.assertEqual(os.path.join(self.config_path, 'foo\\bar'),
nparser.abs_path('foo\\bar'))
def test_filedump(self):
nparser = parser.NginxParser(self.config_path)

6
certbot-nginx/certbot_nginx/tests/tls_sni_01_test.py
View file

@ -1,6 +1,5 @@
"""Tests for certbot_nginx.tls_sni_01"""
import unittest
import shutil
import mock
import six
@ -55,11 +54,6 @@ class TlsSniPerformTest(util.NginxTest):
from certbot_nginx import tls_sni_01
self.sni = tls_sni_01.NginxTlsSni01(config)
def tearDown(self):
shutil.rmtree(self.temp_dir)
shutil.rmtree(self.config_dir)
shutil.rmtree(self.work_dir)
@mock.patch("certbot_nginx.configurator"
".NginxConfigurator.choose_vhosts")
def test_perform(self, mock_choose):

18
certbot-nginx/certbot_nginx/tests/util.py
View file

@ -4,6 +4,8 @@ import os
import pkg_resources
import tempfile
import unittest
import shutil
import warnings
import josepy as jose
import mock
@ -33,6 +35,22 @@ class NginxTest(unittest.TestCase): # pylint: disable=too-few-public-methods
self.rsa512jwk = jose.JWKRSA.load(test_util.load_vector(
"rsa512_key.pem"))
def tearDown(self):
# On Windows we have various files which are not correctly closed at the time of tearDown.
# For know, we log them until a proper file close handling is written.
# Useful for development only, so no warning when we are on a CI process.
def onerror_handler(_, path, excinfo):
"""On error handler"""
if not os.environ.get('APPVEYOR'): # pragma: no cover
message = ('Following error occurred when deleting path {0}'
'during tearDown process: {1}'.format(path, str(excinfo)))
warnings.warn(message)
shutil.rmtree(self.temp_dir, onerror=onerror_handler)
shutil.rmtree(self.config_dir, onerror=onerror_handler)
shutil.rmtree(self.work_dir, onerror=onerror_handler)
shutil.rmtree(self.logs_dir, onerror=onerror_handler)
def get_data_filename(filename):
"""Gets the filename of a test data file."""

4
certbot-nginx/local-oldest-requirements.txt
View file

@ -1,2 +1,2 @@
acme[dev]==0.26.0
certbot[dev]==0.22.0
acme[dev]==0.29.0
-e .[dev]

20
certbot-nginx/setup.py
View file

@ -1,5 +1,7 @@
from setuptools import setup
from setuptools import find_packages
from setuptools.command.test import test as TestCommand
import sys
version = '0.32.0.dev0'
@ -21,6 +23,22 @@ docs_extras = [
'sphinx_rtd_theme',
]
class PyTest(TestCommand):
user_options = []
def initialize_options(self):
TestCommand.initialize_options(self)
self.pytest_args = ''
def run_tests(self):
import shlex
# import here, cause outside the eggs aren't loaded
import pytest
errno = pytest.main(shlex.split(self.pytest_args))
sys.exit(errno)
setup(
name='certbot-nginx',
version=version,
@ -64,4 +82,6 @@ setup(
],
},
test_suite='certbot_nginx',
tests_require=["pytest"],
cmdclass={"test": PyTest},
)

6
certbot/cli.py
View file

@ -108,7 +108,7 @@ manage your account with Let's Encrypt:
# This is the short help for certbot --help, where we disable argparse
# altogether
HELP_USAGE = """
HELP_AND_VERSION_USAGE = """
More detailed help:
-h, --help [TOPIC] print this message, or detailed help on a topic;
@ -117,6 +117,8 @@ More detailed help:
all, automation, commands, paths, security, testing, or any of the
subcommands or plugins (certonly, renew, install, register, nginx,
apache, standalone, webroot, etc.)
--version print the version number
"""
@ -566,7 +568,7 @@ class HelpfulArgumentParser(object):
usage = SHORT_USAGE
if help_arg == True:
self.notify(usage + COMMAND_OVERVIEW % (apache_doc, nginx_doc) + HELP_USAGE)
self.notify(usage + COMMAND_OVERVIEW % (apache_doc, nginx_doc) + HELP_AND_VERSION_USAGE)
sys.exit(0)
elif help_arg in self.COMMANDS_TOPICS:
self.notify(usage + self._list_subcommands())

56
certbot/compat.py
View file

@ -13,13 +13,6 @@ import stat
from certbot import errors
try:
# Linux specific
import fcntl # pylint: disable=import-error
except ImportError:
# Windows specific
import msvcrt # pylint: disable=import-error
UNPRIVILEGED_SUBCOMMANDS_ALLOWED = [
'certificates', 'enhance', 'revoke', 'delete',
'register', 'unregister', 'config_changes', 'plugins']
@ -118,55 +111,6 @@ def readline_with_timeout(timeout, prompt):
return sys.stdin.readline()
def lock_file(fd):
"""
Lock the file linked to the specified file descriptor.
:param int fd: The file descriptor of the file to lock.
"""
if 'fcntl' in sys.modules:
# Linux specific
fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
else:
# Windows specific
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
def release_locked_file(fd, path):
"""
Remove, close, and release a lock file specified by its file descriptor and its path.
:param int fd: The file descriptor of the lock file.
:param str path: The path of the lock file.
"""
# Linux specific
#
# It is important the lock file is removed before it's released,
# otherwise:
#
# process A: open lock file
# process B: release lock file
# process A: lock file
# process A: check device and inode
# process B: delete file
# process C: open and lock a different file at the same path
try:
os.remove(path)
except OSError as err:
if err.errno == errno.EACCES:
# Windows specific
# We will not be able to remove a file before closing it.
# To avoid race conditions described for Linux, we will not delete the lockfile,
# just close it to be reused on the next Certbot call.
pass
else:
raise
finally:
os.close(fd)
def compare_file_modes(mode1, mode2):
"""Return true if the two modes can be considered as equals for this platform"""
if os.name != 'nt':

36
certbot/hooks.py
View file

@ -93,8 +93,7 @@ def _run_pre_hook_if_necessary(command):
if command in executed_pre_hooks:
logger.info("Pre-hook command already run, skipping: %s", command)
else:
logger.info("Running pre-hook command: %s", command)
_run_hook(command)
_run_hook("pre-hook", command)
executed_pre_hooks.add(command)
@ -126,8 +125,7 @@ def post_hook(config):
_run_eventually(cmd)
# certonly / run
elif cmd:
logger.info("Running post-hook command: %s", cmd)
_run_hook(cmd)
_run_hook("post-hook", cmd)
post_hooks = [] # type: List[str]
@ -149,8 +147,7 @@ def _run_eventually(command):
def run_saved_post_hooks():
"""Run any post hooks that were saved up in the course of the 'renew' verb"""
for cmd in post_hooks:
logger.info("Running post-hook command: %s", cmd)
_run_hook(cmd)
_run_hook("post-hook", cmd)
def deploy_hook(config, domains, lineage_path):
@ -220,23 +217,30 @@ def _run_deploy_hook(command, domains, lineage_path, dry_run):
os.environ["RENEWED_DOMAINS"] = " ".join(domains)
os.environ["RENEWED_LINEAGE"] = lineage_path
logger.info("Running deploy-hook command: %s", command)
_run_hook(command)
_run_hook("deploy-hook", command)
def _run_hook(shell_cmd):
def _run_hook(cmd_name, shell_cmd):
"""Run a hook command.
:returns: stderr if there was any"""
:param str cmd_name: the user facing name of the hook being run
:param shell_cmd: shell command to execute
:type shell_cmd: `list` of `str` or `str`
err, _ = execute(shell_cmd)
:returns: stderr if there was any"""
err, _ = execute(cmd_name, shell_cmd)
return err
def execute(shell_cmd):
def execute(cmd_name, shell_cmd):
"""Run a command.
:param str cmd_name: the user facing name of the hook being run
:param shell_cmd: shell command to execute
:type shell_cmd: `list` of `str` or `str`
:returns: `tuple` (`str` stderr, `str` stdout)"""
logger.info("Running %s command: %s", cmd_name, shell_cmd)
# universal_newlines causes Popen.communicate()
# to return str objects instead of bytes in Python 3
@ -245,12 +249,12 @@ def execute(shell_cmd):
out, err = cmd.communicate()
base_cmd = os.path.basename(shell_cmd.split(None, 1)[0])
if out:
logger.info('Output from %s:\n%s', base_cmd, out)
logger.info('Output from %s command %s:\n%s', cmd_name, base_cmd, out)
if cmd.returncode != 0:
logger.error('Hook command "%s" returned error code %d',
shell_cmd, cmd.returncode)
logger.error('%s command "%s" returned error code %d',
cmd_name, shell_cmd, cmd.returncode)
if err:
logger.error('Error output from %s:\n%s', base_cmd, err)
logger.error('Error output from %s command %s:\n%s', cmd_name, base_cmd, err)
return (err, out)

213
certbot/lock.py
View file

@ -1,15 +1,23 @@
"""Implements file locks for locking files and directories in UNIX."""
"""Implements file locks compatible with Linux and Windows for locking files and directories."""
import errno
import logging
import os
try:
import fcntl # pylint: disable=import-error
except ImportError:
import msvcrt # pylint: disable=import-error
POSIX_MODE = False
else:
POSIX_MODE = True
from certbot import compat
from certbot import errors
from acme.magic_typing import Optional, Callable # pylint: disable=unused-import, no-name-in-module
logger = logging.getLogger(__name__)
def lock_dir(dir_path):
# type: (str) -> LockFile
"""Place a lock file on the directory at dir_path.
The lock file is placed in the root of dir_path with the name
@ -27,34 +35,99 @@ def lock_dir(dir_path):
class LockFile(object):
"""A UNIX lock file.
This lock file is released when the locked file is closed or the
process exits. It cannot be used to provide synchronization between
threads. It is based on the lock_file package by Martin Horcicka.
"""
Platform independent file lock system.
LockFile accepts a parameter, the path to a file acting as a lock. Once the LockFile,
instance is created, the associated file is 'locked from the point of view of the OS,
meaning that if another instance of Certbot try at the same time to acquire the same lock,
it will raise an Exception. Calling release method will release the lock, and make it
available to every other instance.
Upon exit, Certbot will also release all the locks.
This allows us to protect a file or directory from being concurrently accessed
or modified by two Certbot instances.
LockFile is platform independent: it will proceed to the appropriate OS lock mechanism
depending on Linux or Windows.
"""
def __init__(self, path):
"""Initialize and acquire the lock file.
:param str path: path to the file to lock
:raises errors.LockError: if unable to acquire the lock
# type: (str) -> None
"""
Create a LockFile instance on the given file path, and acquire lock.
:param str path: the path to the file that will hold a lock
"""
super(LockFile, self).__init__()
self._path = path
self._fd = None
mechanism = _UnixLockMechanism if POSIX_MODE else _WindowsLockMechanism
self._lock_mechanism = mechanism(path)
self.acquire()
def __repr__(self):
# type: () -> str
repr_str = '{0}({1}) <'.format(self.__class__.__name__, self._path)
if self.is_locked():
repr_str += 'acquired>'
else:
repr_str += 'released>'
return repr_str
def acquire(self):
"""Acquire the lock file.
:raises errors.LockError: if lock is already held
:raises OSError: if unable to open or stat the lock file
# type: () -> None
"""
Acquire the lock on the file, forbidding any other Certbot instance to acquire it.
:raises errors.LockError: if unable to acquire the lock
"""
self._lock_mechanism.acquire()
def release(self):
# type: () -> None
"""
Release the lock on the file, allowing any other Certbot instance to acquire it.
"""
self._lock_mechanism.release()
def is_locked(self):
# type: () -> bool
"""
Check if the file is currently locked.
:return: True if the file is locked, False otherwise
"""
return self._lock_mechanism.is_locked()
class _BaseLockMechanism(object):
def __init__(self, path):
# type: (str) -> None
"""
Create a lock file mechanism for Unix.
:param str path: the path to the lock file
"""
self._path = path
self._fd = None # type: Optional[int]
def is_locked(self):
# type: () -> bool
"""Check if lock file is currently locked.
:return: True if the lock file is locked
:rtype: bool
"""
return self._fd is not None
def acquire(self): # pylint: disable=missing-docstring
pass # pragma: no cover
def release(self): # pylint: disable=missing-docstring
pass # pragma: no cover
class _UnixLockMechanism(_BaseLockMechanism):
"""
A UNIX lock file mechanism.
This lock file is released when the locked file is closed or the
process exits. It cannot be used to provide synchronization between
threads. It is based on the lock_file package by Martin Horcicka.
"""
def acquire(self):
# type: () -> None
"""Acquire the lock."""
while self._fd is None:
# Open the file
fd = os.open(self._path, os.O_CREAT | os.O_WRONLY, 0o600)
@ -68,33 +141,29 @@ class LockFile(object):
os.close(fd)
def _try_lock(self, fd):
"""Try to acquire the lock file without blocking.
# type: (int) -> None
"""
Try to acquire the lock file without blocking.
:param int fd: file descriptor of the opened file to lock
"""
try:
compat.lock_file(fd)
fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError as err:
if err.errno in (errno.EACCES, errno.EAGAIN):
logger.debug(
"A lock on %s is held by another process.", self._path)
raise errors.LockError(
"Another instance of Certbot is already running.")
logger.debug('A lock on %s is held by another process.', self._path)
raise errors.LockError('Another instance of Certbot is already running.')
raise
def _lock_success(self, fd):
"""Did we successfully grab the lock?
# type: (int) -> bool
"""
Did we successfully grab the lock?
Because this class deletes the locked file when the lock is
released, it is possible another process removed and recreated
the file between us opening the file and acquiring the lock.
:param int fd: file descriptor of the opened file to lock
:returns: True if the lock was successfully acquired
:rtype: bool
"""
try:
stat1 = os.stat(self._path)
@ -108,17 +177,75 @@ class LockFile(object):
# the same device and inode, they're the same file.
return stat1.st_dev == stat2.st_dev and stat1.st_ino == stat2.st_ino
def __repr__(self):
repr_str = '{0}({1}) <'.format(self.__class__.__name__, self._path)
if self._fd is None:
repr_str += 'released>'
else:
repr_str += 'acquired>'
return repr_str
def release(self):
# type: () -> None
"""Remove, close, and release the lock file."""
# It is important the lock file is removed before it's released,
# otherwise:
#
# process A: open lock file
# process B: release lock file
# process A: lock file
# process A: check device and inode
# process B: delete file
# process C: open and lock a different file at the same path
try:
os.remove(self._path)
finally:
# Following check is done to make mypy happy: it ensure that self._fd, marked
# as Optional[int] is effectively int to make it compatible with os.close signature.
if self._fd is None: # pragma: no cover
raise TypeError('Error, self._fd is None.')
try:
os.close(self._fd)
finally:
self._fd = None
class _WindowsLockMechanism(_BaseLockMechanism):
"""
A Windows lock file mechanism.
By default on Windows, acquiring a file handler gives exclusive access to the process
and results in an effective lock. However, it is possible to explicitly acquire the
file handler in shared access in terms of read and write, and this is done by os.open
and io.open in Python. So an explicit lock needs to be done through the call of
msvcrt.locking, that will lock the first byte of the file. In theory, it is also
possible to access a file in shared delete access, allowing other processes to delete an
opened file. But this needs also to be done explicitly by all processes using the Windows
low level APIs, and Python does not do it. As of Python 3.7 and below, Python developers
state that deleting a file opened by a process from another process is not possible with
os.open and io.open.
Consequently, mscvrt.locking is sufficient to obtain an effective lock, and the race
condition encountered on Linux is not possible on Windows, leading to a simpler workflow.
"""
def acquire(self):
"""Acquire the lock"""
open_mode = os.O_RDWR | os.O_CREAT | os.O_TRUNC
fd = os.open(self._path, open_mode, 0o600)
try:
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
except (IOError, OSError) as err:
os.close(fd)
# Anything except EACCES is unexpected. Raise directly the error in that case.
if err.errno != errno.EACCES:
raise
logger.debug('A lock on %s is held by another process.', self._path)
raise errors.LockError('Another instance of Certbot is already running.')
self._fd = fd
def release(self):
"""Remove, close, and release the lock file."""
"""Release the lock."""
try:
compat.release_locked_file(self._fd, self._path)
msvcrt.locking(self._fd, msvcrt.LK_UNLCK, 1)
os.close(self._fd)
try:
os.remove(self._path)
except OSError as e:
# If the lock file cannot be removed, it is not a big deal.
# Likely another instance is acquiring the lock we just released.
logger.debug(str(e))
finally:
self._fd = None

7
certbot/plugins/manual.py
View file

@ -202,7 +202,7 @@ permitted by DNS standards.)
os.environ.pop('CERTBOT_KEY_PATH', None)
os.environ.pop('CERTBOT_SNI_DOMAIN', None)
os.environ.update(env)
_, out = hooks.execute(self.conf('auth-hook'))
_, out = self._execute_hook('auth-hook')
env['CERTBOT_AUTH_OUTPUT'] = out.strip()
self.env[achall] = env
@ -243,5 +243,8 @@ permitted by DNS standards.)
if 'CERTBOT_TOKEN' not in env:
os.environ.pop('CERTBOT_TOKEN', None)
os.environ.update(env)
hooks.execute(self.conf('cleanup-hook'))
self._execute_hook('cleanup-hook')
self.reverter.recovery_routine()
def _execute_hook(self, hook_name):
return hooks.execute(self.option_name(hook_name), self.conf(hook_name))

34
certbot/tests/hook_test.py
View file

@ -121,7 +121,7 @@ class PreHookTest(HookTest):
def _test_nonrenew_common(self):
mock_execute = self._call_with_mock_execute(self.config)
mock_execute.assert_called_once_with(self.config.pre_hook)
mock_execute.assert_called_once_with("pre-hook", self.config.pre_hook)
self._test_no_executions_common()
def test_no_hooks(self):
@ -137,21 +137,21 @@ class PreHookTest(HookTest):
def test_renew_disabled_dir_hooks(self):
self.config.directory_hooks = False
mock_execute = self._call_with_mock_execute(self.config)
mock_execute.assert_called_once_with(self.config.pre_hook)
mock_execute.assert_called_once_with("pre-hook", self.config.pre_hook)
self._test_no_executions_common()
def test_renew_no_overlap(self):
self.config.verb = "renew"
mock_execute = self._call_with_mock_execute(self.config)
mock_execute.assert_any_call(self.dir_hook)
mock_execute.assert_called_with(self.config.pre_hook)
mock_execute.assert_any_call("pre-hook", self.dir_hook)
mock_execute.assert_called_with("pre-hook", self.config.pre_hook)
self._test_no_executions_common()
def test_renew_with_overlap(self):
self.config.pre_hook = self.dir_hook
self.config.verb = "renew"
mock_execute = self._call_with_mock_execute(self.config)
mock_execute.assert_called_once_with(self.dir_hook)
mock_execute.assert_called_once_with("pre-hook", self.dir_hook)
self._test_no_executions_common()
def _test_no_executions_common(self):
@ -193,7 +193,7 @@ class PostHookTest(HookTest):
for verb in ("certonly", "run",):
self.config.verb = verb
mock_execute = self._call_with_mock_execute(self.config)
mock_execute.assert_called_once_with(self.config.post_hook)
mock_execute.assert_called_once_with("post-hook", self.config.post_hook)
self.assertFalse(self._get_eventually())
def test_cert_only_and_run_without_hook(self):
@ -277,12 +277,12 @@ class RunSavedPostHooksTest(HookTest):
calls = mock_execute.call_args_list
for actual_call, expected_arg in zip(calls, self.eventually):
self.assertEqual(actual_call[0][0], expected_arg)
self.assertEqual(actual_call[0][1], expected_arg)
def test_single(self):
self.eventually = ["foo"]
mock_execute = self._call_with_mock_execute_and_eventually()
mock_execute.assert_called_once_with(self.eventually[0])
mock_execute.assert_called_once_with("post-hook", self.eventually[0])
class RenewalHookTest(HookTest):
@ -360,7 +360,7 @@ class DeployHookTest(RenewalHookTest):
self.config.deploy_hook = "foo"
mock_execute = self._call_with_mock_execute(
self.config, domains, lineage)
mock_execute.assert_called_once_with(self.config.deploy_hook)
mock_execute.assert_called_once_with("deploy-hook", self.config.deploy_hook)
class RenewHookTest(RenewalHookTest):
@ -384,7 +384,7 @@ class RenewHookTest(RenewalHookTest):
self.config.directory_hooks = False
mock_execute = self._call_with_mock_execute(
self.config, ["example.org"], "/foo/bar")
mock_execute.assert_called_once_with(self.config.renew_hook)
mock_execute.assert_called_once_with("deploy-hook", self.config.renew_hook)
@mock.patch("certbot.hooks.logger")
def test_dry_run(self, mock_logger):
@ -408,13 +408,13 @@ class RenewHookTest(RenewalHookTest):
self.config.renew_hook = self.dir_hook
mock_execute = self._call_with_mock_execute(
self.config, ["example.net", "example.org"], "/foo/bar")
mock_execute.assert_called_once_with(self.dir_hook)
mock_execute.assert_called_once_with("deploy-hook", self.dir_hook)
def test_no_overlap(self):
mock_execute = self._call_with_mock_execute(
self.config, ["example.org"], "/foo/bar")
mock_execute.assert_any_call(self.dir_hook)
mock_execute.assert_called_with(self.config.renew_hook)
mock_execute.assert_any_call("deploy-hook", self.dir_hook)
mock_execute.assert_called_with("deploy-hook", self.config.renew_hook)
class ExecuteTest(unittest.TestCase):
@ -433,18 +433,22 @@ class ExecuteTest(unittest.TestCase):
def _test_common(self, returncode, stdout, stderr):
given_command = "foo"
given_name = "foo-hook"
with mock.patch("certbot.hooks.Popen") as mock_popen:
mock_popen.return_value.communicate.return_value = (stdout, stderr)
mock_popen.return_value.returncode = returncode
with mock.patch("certbot.hooks.logger") as mock_logger:
self.assertEqual(self._call(given_command), (stderr, stdout))
self.assertEqual(self._call(given_name, given_command), (stderr, stdout))
executed_command = mock_popen.call_args[1].get(
"args", mock_popen.call_args[0][0])
self.assertEqual(executed_command, given_command)
mock_logger.info.assert_any_call("Running %s command: %s",
given_name, given_command)
if stdout:
self.assertTrue(mock_logger.info.called)
mock_logger.info.assert_any_call(mock.ANY, mock.ANY,
mock.ANY, stdout)
if stderr or returncode:
self.assertTrue(mock_logger.error.called)

58
certbot/tests/lock_test.py
View file

@ -3,6 +3,12 @@ import functools
import multiprocessing
import os
import unittest
try:
import fcntl # pylint: disable=import-error,unused-import
except ImportError:
POSIX_MODE = False
else:
POSIX_MODE = True
import mock
@ -10,7 +16,6 @@ from certbot import errors
from certbot.tests import util as test_util
@test_util.broken_on_windows
class LockDirTest(test_util.TempDirTestCase):
"""Tests for certbot.lock.lock_dir."""
@classmethod
@ -25,7 +30,6 @@ class LockDirTest(test_util.TempDirTestCase):
test_util.lock_and_call(assert_raises, lock_path)
@test_util.broken_on_windows
class LockFileTest(test_util.TempDirTestCase):
"""Tests for certbot.lock.LockFile."""
@classmethod
@ -37,6 +41,7 @@ class LockFileTest(test_util.TempDirTestCase):
super(LockFileTest, self).setUp()
self.lock_path = os.path.join(self.tempdir, 'test.lock')
@test_util.broken_on_windows
def test_acquire_without_deletion(self):
# acquire the lock in another process but don't delete the file
child = multiprocessing.Process(target=self._call,
@ -54,6 +59,7 @@ class LockFileTest(test_util.TempDirTestCase):
self.assertRaises, errors.LockError, self._call, self.lock_path)
test_util.lock_and_call(assert_raises, self.lock_path)
@test_util.broken_on_windows
def test_locked_repr(self):
lock_file = self._call(self.lock_path)
locked_repr = repr(lock_file)
@ -71,6 +77,8 @@ class LockFileTest(test_util.TempDirTestCase):
self.assertTrue(lock_file.__class__.__name__ in lock_repr)
self.assertTrue(self.lock_path in lock_repr)
@test_util.skip_on_windows(
'Race conditions on lock are specific to the non-blocking file access approach on Linux.')
def test_race(self):
should_delete = [True, False]
stat = os.stat
@ -86,32 +94,42 @@ class LockFileTest(test_util.TempDirTestCase):
self._call(self.lock_path)
self.assertFalse(should_delete)
@test_util.broken_on_windows
def test_removed(self):
lock_file = self._call(self.lock_path)
lock_file.release()
self.assertFalse(os.path.exists(self.lock_path))
@mock.patch('certbot.compat.fcntl.lockf')
def test_unexpected_lockf_err(self, mock_lockf):
def test_unexpected_lockf_or_locking_err(self):
if POSIX_MODE:
mocked_function = 'certbot.lock.fcntl.lockf'
else:
mocked_function = 'certbot.lock.msvcrt.locking'
msg = 'hi there'
mock_lockf.side_effect = IOError(msg)
try:
self._call(self.lock_path)
except IOError as err:
self.assertTrue(msg in str(err))
else: # pragma: no cover
self.fail('IOError not raised')
with mock.patch(mocked_function) as mock_lock:
mock_lock.side_effect = IOError(msg)
try:
self._call(self.lock_path)
except IOError as err:
self.assertTrue(msg in str(err))
else: # pragma: no cover
self.fail('IOError not raised')
@mock.patch('certbot.lock.os.stat')
def test_unexpected_stat_err(self, mock_stat):
def test_unexpected_os_err(self):
if POSIX_MODE:
mock_function = 'certbot.lock.os.stat'
else:
mock_function = 'certbot.lock.msvcrt.locking'
# The only expected errno are ENOENT and EACCES in lock module.
msg = 'hi there'
mock_stat.side_effect = OSError(msg)
try:
self._call(self.lock_path)
except OSError as err:
self.assertTrue(msg in str(err))
else: # pragma: no cover
self.fail('OSError not raised')
with mock.patch(mock_function) as mock_os:
mock_os.side_effect = OSError(msg)
try:
self._call(self.lock_path)
except OSError as err:
self.assertTrue(msg in str(err))
else: # pragma: no cover
self.fail('OSError not raised')
if __name__ == "__main__":

117
certbot/tests/util.py
View file

@ -3,14 +3,15 @@
.. warning:: This module is not part of the public API.
"""
import multiprocessing
import logging
import os
import pkg_resources
import shutil
import stat
import tempfile
import unittest
import sys
import warnings
from multiprocessing import Process, Event
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
@ -23,8 +24,9 @@ from six.moves import reload_module # pylint: disable=import-error
from certbot import constants
from certbot import interfaces
from certbot import storage
from certbot import util
from certbot import configuration
from certbot import lock
from certbot import util
from certbot.display import util as display_util
@ -211,7 +213,7 @@ class FreezableMock(object):
"""
def __init__(self, frozen=False, func=None, return_value=mock.sentinel.DEFAULT):
self._frozen_set = set() if frozen else set(('freeze',))
self._frozen_set = set() if frozen else {'freeze', }
self._func = func
self._mock = mock.MagicMock()
if return_value != mock.sentinel.DEFAULT:
@ -328,22 +330,25 @@ class TempDirTestCase(unittest.TestCase):
def tearDown(self):
"""Execute after test"""
# On Windows we have various files which are not correctly closed at the time of tearDown.
# For know, we log them until a proper file close handling is written.
# Useful for development only, so no warning when we are on a CI process.
def onerror_handler(_, path, excinfo):
"""On error handler"""
if not os.environ.get('APPVEYOR'): # pragma: no cover
message = ('Following error occurred when deleting the tempdir {0}'
' for path {1} during tearDown process: {2}'
.format(self.tempdir, path, str(excinfo)))
warnings.warn(message)
shutil.rmtree(self.tempdir, onerror=onerror_handler)
# Cleanup opened resources after a test. This is usually done through atexit handlers in
# Certbot, but during tests, atexit will not run registered functions before tearDown is
# called and instead will run them right before the entire test process exits.
# It is a problem on Windows, that does not accept to clean resources before closing them.
logging.shutdown()
# Remove logging handlers that have been closed so they won't be
# accidentally used in future tests.
logging.getLogger().handlers = []
util._release_locks() # pylint: disable=protected-access
def handle_rw_files(_, path, __):
"""Handle read-only files, that will fail to be removed on Windows."""
os.chmod(path, stat.S_IWRITE)
os.remove(path)
shutil.rmtree(self.tempdir, onerror=handle_rw_files)
class ConfigTestCase(TempDirTestCase):
"""Test class which sets up a NamespaceConfig object.
"""
"""Test class which sets up a NamespaceConfig object."""
def setUp(self):
super(ConfigTestCase, self).setUp()
self.config = configuration.NamespaceConfig(
@ -358,47 +363,51 @@ class ConfigTestCase(TempDirTestCase):
self.config.chain_path = constants.CLI_DEFAULTS['auth_chain_path']
self.config.server = "https://example.com"
def lock_and_call(func, lock_path):
"""Grab a lock for lock_path and call func.
:param callable func: object to call after acquiring the lock
:param str lock_path: path to file or directory to lock
def _handle_lock(event_in, event_out, path):
"""
# Reload module to reset internal _LOCKS dictionary
Acquire a file lock on given path, then wait to release it. This worker is coordinated
using events to signal when the lock should be acquired and released.
:param multiprocessing.Event event_in: event object to signal when to release the lock
:param multiprocessing.Event event_out: event object to signal when the lock is acquired
:param path: the path to lock
"""
if os.path.isdir(path):
my_lock = lock.lock_dir(path)
else:
my_lock = lock.LockFile(path)
try:
event_out.set()
assert event_in.wait(timeout=20), 'Timeout while waiting to release the lock.'
finally:
my_lock.release()
def lock_and_call(callback, path_to_lock):
"""
Grab a lock on path_to_lock from a foreign process then execute the callback.
:param callable callback: object to call after acquiring the lock
:param str path_to_lock: path to file or directory to lock
"""
# Reload certbot.util module to reset internal _LOCKS dictionary.
reload_module(util)
# start child and wait for it to grab the lock
cv = multiprocessing.Condition()
cv.acquire()
child_args = (cv, lock_path,)
child = multiprocessing.Process(target=hold_lock, args=child_args)
child.start()
cv.wait()
emit_event = Event()
receive_event = Event()
process = Process(target=_handle_lock, args=(emit_event, receive_event, path_to_lock))
process.start()
# call func and terminate the child
func()
cv.notify()
cv.release()
child.join()
assert child.exitcode == 0
# Wait confirmation that lock is acquired
assert receive_event.wait(timeout=10), 'Timeout while waiting to acquire the lock.'
# Execute the callback
callback()
# Trigger unlock from foreign process
emit_event.set()
def hold_lock(cv, lock_path): # pragma: no cover
"""Acquire a file lock at lock_path and wait to release it.
# Wait for process termination
process.join(timeout=10)
assert process.exitcode == 0
:param multiprocessing.Condition cv: condition for synchronization
:param str lock_path: path to the file lock
"""
from certbot import lock
if os.path.isdir(lock_path):
my_lock = lock.lock_dir(lock_path)
else:
my_lock = lock.LockFile(lock_path)
cv.acquire()
cv.notify()
cv.wait()
my_lock.release()
def skip_on_windows(reason):
"""Decorator to skip permanently a test on Windows. A reason is required."""
@ -407,6 +416,7 @@ def skip_on_windows(reason):
return unittest.skipIf(sys.platform == 'win32', reason)(function)
return wrapper
def broken_on_windows(function):
"""Decorator to skip temporarily a broken test on Windows."""
reason = 'Test is broken and ignored on windows but should be fixed.'
@ -415,9 +425,10 @@ def broken_on_windows(function):
and os.environ.get('SKIP_BROKEN_TESTS_ON_WINDOWS', 'true') == 'true',
reason)(function)
def temp_join(path):
"""
Return the given path joined to the tempdir path for the current platform
Eg.: 'cert' => /tmp/cert (Linux) or 'C:\\Users\\currentuser\\AppData\\Temp\\cert' (Windows)
"""
return os.path.join(tempfile.gettempdir(), path)
return os.path.join(tempfile.gettempdir(), path)

40
certbot/tests/util_test.py
View file

@ -2,7 +2,6 @@
import argparse
import errno
import os
import shutil
import unittest
import mock
@ -88,7 +87,6 @@ class LockDirUntilExit(test_util.TempDirTestCase):
import certbot.util
reload_module(certbot.util)
@test_util.broken_on_windows
@mock.patch('certbot.util.logger')
@mock.patch('certbot.util.atexit_register')
def test_it(self, mock_register, mock_logger):
@ -100,11 +98,15 @@ class LockDirUntilExit(test_util.TempDirTestCase):
self.assertEqual(mock_register.call_count, 1)
registered_func = mock_register.call_args[0][0]
shutil.rmtree(subdir)
registered_func() # exception not raised
# logger.debug is only called once because the second call
# to lock subdir was ignored because it was already locked
self.assertEqual(mock_logger.debug.call_count, 1)
from certbot import util
# Despite lock_dir_until_exit has been called twice to subdir, its lock should have been
# added only once. So we expect to have two lock references: for self.tempdir and subdir
self.assertTrue(len(util._LOCKS) == 2) # pylint: disable=protected-access
registered_func() # Exception should not be raised
# Logically, logger.debug, that would be invoked in case of unlock failure,
# should never been called.
self.assertEqual(mock_logger.debug.call_count, 0)
class SetUpCoreDirTest(test_util.TempDirTestCase):
@ -191,7 +193,12 @@ class CheckPermissionsTest(test_util.TempDirTestCase):
def test_wrong_mode(self):
os.chmod(self.tempdir, 0o400)
self.assertFalse(self._call(0o600))
try:
self.assertFalse(self._call(0o600))
finally:
# Without proper write permissions, Windows is unable to delete a folder,
# even with admin permissions. Write access must be explicitly set first.
os.chmod(self.tempdir, 0o700)
class UniqueFileTest(test_util.TempDirTestCase):
@ -277,20 +284,9 @@ class UniqueLineageNameTest(test_util.TempDirTestCase):
for f, _ in items:
f.close()
@mock.patch("certbot.util.os.fdopen")
def test_failure(self, mock_fdopen):
err = OSError("whoops")
err.errno = errno.EIO
mock_fdopen.side_effect = err
self.assertRaises(OSError, self._call, "wow")
@mock.patch("certbot.util.os.fdopen")
def test_subsequent_failure(self, mock_fdopen):
self._call("wow")
err = OSError("whoops")
err.errno = errno.EIO
mock_fdopen.side_effect = err
self.assertRaises(OSError, self._call, "wow")
def test_failure(self):
with mock.patch("certbot.util.os.open", side_effect=OSError(errno.EIO)):
self.assertRaises(OSError, self._call, "wow")
class SafelyRemoveTest(test_util.TempDirTestCase):

6
certbot/util.py
View file

@ -142,6 +142,7 @@ def _release_locks():
except: # pylint: disable=bare-except
msg = 'Exception occurred releasing lock: {0!r}'.format(dir_lock)
logger.debug(msg, exc_info=True)
_LOCKS.clear()
def set_up_core_dir(directory, mode, uid, strict):
@ -225,9 +226,8 @@ def safe_open(path, mode="w", chmod=None, buffering=None):
fdopen_args = () # type: Union[Tuple[()], Tuple[int]]
if buffering is not None:
fdopen_args = (buffering,)
return os.fdopen(
os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, *open_args),
mode, *fdopen_args)
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, *open_args)
return os.fdopen(fd, mode, *fdopen_args)
def _unique_file(path, filename_pat, count, chmod, mode):

20
setup.py
View file

@ -1,8 +1,10 @@
import codecs
import os
import re
import sys
from setuptools import find_packages, setup
from setuptools.command.test import test as TestCommand
# Workaround for http://bugs.python.org/issue8876, see
# http://bugs.python.org/issue8876#msg208792
@ -77,6 +79,22 @@ docs_extras = [
'sphinx_rtd_theme',
]
class PyTest(TestCommand):
user_options = []
def initialize_options(self):
TestCommand.initialize_options(self)
self.pytest_args = ''
def run_tests(self):
import shlex
# import here, cause outside the eggs aren't loaded
import pytest
errno = pytest.main(shlex.split(self.pytest_args))
sys.exit(errno)
setup(
name='certbot',
version=version,
@ -123,6 +141,8 @@ setup(
# to test all packages run "python setup.py test -s
# {acme,certbot_apache,certbot_nginx}"
test_suite='certbot',
tests_require=["pytest"],
cmdclass={"test": PyTest},
entry_points={
'console_scripts': [

22
tests/display.py
View file

@ -1,22 +0,0 @@
"""Manual test of display functions."""
import sys
from certbot.display import util
from certbot.tests.display import util_test
def test_visual(displayer, choices):
"""Visually test all of the display functions."""
displayer.notification("Random notification!")
displayer.menu("Question?", choices,
ok_label="O", cancel_label="Can", help_label="??")
displayer.menu("Question?", [choice[1] for choice in choices],
ok_label="O", cancel_label="Can", help_label="??")
displayer.input("Input Message")
displayer.yesno("YesNo Message", yes_label="Yessir", no_label="Nosir")
displayer.checklist("Checklist Message", [choice[0] for choice in choices])
if __name__ == "__main__":
displayer = util.FileDisplay(sys.stdout, False)
test_visual(displayer, util_test.CHOICES)

5
tests/letstest/multitester.py
View file

@ -564,6 +564,11 @@ try:
ii, target, status = outq
print('%d %s %s'%(ii, target['name'], status))
results_file.write('%d %s %s\n'%(ii, target['name'], status))
if len(outputs) != num_processes:
failure_message = 'FAILURE: Some target machines failed to run and were not tested. ' +\
'Tests should be rerun.'
print(failure_message)
results_file.write(failure_message + '\n')
results_file.close()
finally:

2
tests/letstest/scripts/test_sdists.sh
View file

@ -12,6 +12,8 @@ export VENV_ARGS="-p $PYTHON"
# setup venv
tools/_venv_common.py --requirement letsencrypt-auto-source/pieces/dependency-requirements.txt
. ./venv/bin/activate
# pytest is needed to run tests on some of our packages so we install a pinned version here.
tools/pip_install.py pytest
# build sdists
for pkg_dir in acme . $PLUGINS; do

6
tests/letstest/scripts/test_tests.sh
View file

@ -1,18 +1,20 @@
#!/bin/sh -xe
LE_AUTO="letsencrypt/letsencrypt-auto-source/letsencrypt-auto"
REPO_ROOT="letsencrypt"
LE_AUTO="$REPO_ROOT/letsencrypt-auto-source/letsencrypt-auto"
LE_AUTO="$LE_AUTO --debug --no-self-upgrade --non-interactive"
MODULES="acme certbot certbot_apache certbot_nginx"
PIP_INSTALL="$REPO_ROOT/tools/pip_install.py"
VENV_NAME=venv
# *-auto respects VENV_PATH
$LE_AUTO --os-packages-only
LE_AUTO_SUDO="" VENV_PATH="$VENV_NAME" $LE_AUTO --no-bootstrap --version
. $VENV_NAME/bin/activate
"$PIP_INSTALL" pytest
# change to an empty directory to ensure CWD doesn't affect tests
cd $(mktemp -d)
pip install pytest==3.2.5
for module in $MODULES ; do
echo testing $module

2
tools/install_and_test.py
View file

@ -15,7 +15,7 @@ import subprocess
import re
SKIP_PROJECTS_ON_WINDOWS = [
'certbot-apache', 'certbot-nginx', 'certbot-postfix', 'letshelp-certbot']
'certbot-apache', 'certbot-postfix', 'letshelp-certbot']
def call_with_print(command, cwd=None):

5
tox.cover.py
View file

@ -35,7 +35,8 @@ COVER_THRESHOLDS = {
}
SKIP_PROJECTS_ON_WINDOWS = [
'certbot-apache', 'certbot-nginx', 'certbot-postfix', 'letshelp-certbot']
'certbot-apache', 'certbot-postfix', 'letshelp-certbot']
def cover(package):
threshold = COVER_THRESHOLDS.get(package)['windows' if os.name == 'nt' else 'linux']
@ -54,6 +55,7 @@ def cover(package):
sys.executable, '-m', 'coverage', 'report', '--fail-under', str(threshold), '--include',
'{0}/*'.format(pkg_dir), '--show-missing'])
def main():
description = """
This script is used by tox.ini (and thus by Travis CI and AppVeyor) in order
@ -77,5 +79,6 @@ Option -e makes sure we fail fast and don't submit to codecov."""
for package in packages:
cover(package)
if __name__ == '__main__':
main()