From b666f5e4ed745d7901f44f5f90e97d74e5277132 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 11 Oct 2019 10:16:33 +0200 Subject: [PATCH] Refactor Patroni REST API communication (#1197) * make it possible to use client certificates with REST API * define a separate PatroniRequest class which handles all communication * refactor patronictl to use the new class * make Ha to use the new class instead of calling requests.get. The old call wasn't taking into account certificates and basic-auth Close #898 --- docs/ENVIRONMENT.rst | 9 +++ docs/SETTINGS.rst | 10 ++- patroni/__init__.py | 4 + patroni/api.py | 103 +++++++++++++----------- patroni/config.py | 3 +- patroni/ctl.py | 90 ++++++++------------- patroni/ha.py | 10 +-- patroni/request.py | 47 +++++++++++ tests/test_api.py | 10 ++- tests/test_ctl.py | 186 +++++++++++++++++++------------------------ tests/test_ha.py | 9 +-- 11 files changed, 253 insertions(+), 228 deletions(-) create mode 100644 patroni/request.py diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index f3fedd67..7b3517da 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -117,6 +117,15 @@ REST API - **PATRONI\_RESTAPI\_PASSWORD**: Basic-auth password to protect unsafe REST API endpoints. - **PATRONI\_RESTAPI\_CERTFILE**: Specifies the file with the certificate in the PEM format. If the certfile is not specified or is left empty, the API server will work without SSL. - **PATRONI\_RESTAPI\_KEYFILE**: Specifies the file with the secret key in the PEM format. +- **PATRONI\_RESTAPI\_CAFILE**: Specifies the file with the CA_BUNDLE with certificates of trusted CAs to use while verifying client certs. +- **PATRONI\_RESTAPI\_VERIFY\_CLIENT**: ``none``, ``optional`` or ``required``. When ``none`` REST API will not check client certificates. When ``required`` client certificates are required for all REST API calls. When ``optional`` client certificates are required for all unsafe REST API endpoints. If ``verify_client`` is set to ``optional`` or ``required`` basic-auth is not checked. + +CTL +--- +- **PATRONI\_CTL\_INSECURE**: Allow connections to REST API without verifying SSL certs. +- **PATRONI\_CTL\_CACERT**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter. +- **PATRONI\_CTL\_CERTFILE**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter. +- **PATRONI\_CTL\_KEYFILE**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter. ZooKeeper --------- diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 1e22b57c..139696c3 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -152,12 +152,12 @@ PostgreSQL - **rewind**: - **username**: name for the user for ``pg_rewind``; the user will be created during initialization of postgres 11+ and all necessary `permissions `__ will be granted. - **password**: password for the user for ``pg_rewind``; the user will be created during initialization. -- **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.) - **sslmode**: (optional) maps to the `sslmode `__ connection parameter, which allows a client to specify the type of TLS negotiation mode with the server. For more information on how each mode works, please visit the `PostgreSQL documentation `__. The default mode is ``prefer``. - **sslkey**: (optional) maps to the `sslkey `__ connection parameter, which specifies the location of the secret key used with the client's certificate. - **sslcert**: (optional) maps to the `sslcert `__ connection parameter, which specifies the location of the client certificate. - **sslrootcert**: (optional) maps to the `sslrootcert `__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate. - **sslcrl**: (optional) maps to the `sslcrl `__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list. +- **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.) - **on\_reload**: run this script when configuration reload is triggered. - **on\_restart**: run this script when the postgres restarts (without changing role). - **on\_role\_change**: run this script when the postgres is being promoted or demoted. @@ -199,9 +199,10 @@ REST API - **username**: Basic-auth username to protect unsafe REST API endpoints. - **password**: Basic-auth password to protect unsafe REST API endpoints. - - **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. - **certfile**: Specifies the file with the certificate in the PEM format. If the certfile is not specified or is left empty, the API server will work without SSL. - **keyfile**: Specifies the file with the secret key in the PEM format. + - **cafile**: Specifies the file with the CA_BUNDLE with certificates of trusted CAs to use while verifying client certs. + - **verify\_client**: ``none``, ``optional`` or ``required``. When ``none`` REST API will not check client certificates. When ``required`` client certificates are required for all REST API calls. When ``optional`` client certificates are required for all unsafe REST API endpoints. If ``verify_client`` is set to ``optional`` or ``required`` basic-auth is not checked. .. _patronictl_settings: @@ -209,8 +210,9 @@ CTL --- - **Optional**: - **insecure**: Allow connections to REST API without verifying SSL certs. - - **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cacert" parameter. - - **certfile**: Specifies the file with the certificate in the PEM format to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "certfile" parameter. + - **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter. + - **certfile**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter. + - **keyfile**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter. ZooKeeper ---------- diff --git a/patroni/__init__.py b/patroni/__init__.py index 37802a33..a9c99835 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -16,6 +16,7 @@ class Patroni(object): from patroni.ha import Ha from patroni.log import PatroniLogger from patroni.postgresql import Postgresql + from patroni.request import PatroniRequest from patroni.version import __version__ from patroni.watchdog import Watchdog @@ -31,6 +32,7 @@ class Patroni(object): self.postgresql = Postgresql(self.config['postgresql']) self.api = RestApiServer(self, self.config['restapi']) + self.request = PatroniRequest(self.config, True) self.ha = Ha(self) self.tags = self.get_tags() @@ -71,6 +73,8 @@ class Patroni(object): self.tags = self.get_tags() self.logger.reload_config(self.config.get('log', {})) self.watchdog.reload_config(self.config) + if sighup: + self.request.reload_config(self.config) self.api.reload_config(self.config['restapi']) self.postgresql.reload_config(self.config['postgresql'], sighup) self.dcs.reload_config(self.config) diff --git a/patroni/api.py b/patroni/api.py index b8e77b5b..b14347c4 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -7,6 +7,7 @@ import traceback import dateutil.parser import datetime import os +import six import socket from patroni.postgresql import PostgresConnectionException @@ -20,20 +21,6 @@ from threading import Thread logger = logging.getLogger(__name__) -def check_auth(func): - """Decorator function to check authorization header. - - Usage example: - @check_auth - def do_PUT_foo(): - pass - """ - def wrapper(handler, *args, **kwargs): - if handler.check_auth_header(): - return func(handler, *args, **kwargs) - return wrapper - - class RestApiHandler(BaseHTTPRequestHandler): def _write_response(self, status_code, body, content_type='text/html', headers=None): @@ -49,14 +36,20 @@ class RestApiHandler(BaseHTTPRequestHandler): def _write_json_response(self, status_code, response): self._write_response(status_code, json.dumps(response), content_type='application/json') - def send_auth_request(self, body): - headers = {'WWW-Authenticate': 'Basic realm="' + self.server.patroni.__class__.__name__ + '"'} - self._write_response(401, body, headers=headers) + def check_auth(func): + """Decorator function to check authorization header or client certificates - def check_auth_header(self): - auth_header = self.headers.get('Authorization') - status = self.server.check_auth_header(auth_header) - return not status or self.send_auth_request(status) + Usage example: + @check_auth + def do_PUT_foo(): + pass + """ + + def wrapper(self, *args, **kwargs): + if self.server.check_auth(self): + return func(self, *args, **kwargs) + + return wrapper def _write_status_response(self, status_code, response): patroni = self.server.patroni @@ -488,8 +481,8 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): def __init__(self, patroni, config): self.patroni = patroni self.__listen = None - self.__initialize(config) - self.__set_config_parameters(config) + self.__ssl_options = None + self.reload_config(config) self.daemon = True def query(self, sql, *params): @@ -520,13 +513,16 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): if not auth_header.startswith('Basic ') or not self.check_basic_auth_key(auth_header[6:]): return 'not authenticated' - @staticmethod - def __get_ssl_options(config): - return {option: config[option] for option in ['certfile', 'keyfile'] if option in config} + def check_auth(self, rh): + if not hasattr(rh.request, 'getpeercert') or not rh.request.getpeercert(): # valid client cert isn't present + if self.__protocol == 'https' and self.__ssl_options.get('verify_client') in ('required', 'optional'): + return rh._write_response(403, 'client certificate required') - def __set_config_parameters(self, config): - self.__auth_key = base64.b64encode(config['auth'].encode('utf-8')).decode('utf-8') if 'auth' in config else None - self.connection_string = uri(self.__protocol, config.get('connect_address') or self.__listen, 'patroni') + reason = self.check_auth_header(rh.headers.get('Authorization')) + if reason: + headers = {'WWW-Authenticate': 'Basic realm="' + self.patroni.__class__.__name__ + '"'} + return rh._write_response(401, reason, headers=headers) + return True @staticmethod def __has_dual_stack(): @@ -545,7 +541,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): def __httpserver_init(self, host, port): dual_stack = self.__has_dual_stack() - if host == '': + if host in ('', '*'): host = None info = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_PASSIVE) @@ -555,43 +551,56 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): self.address_family = info[0][0] HTTPServer.__init__(self, info[0][-1][:2], RestApiHandler) - def __initialize(self, config): + def __initialize(self, listen, ssl_options): try: - host, port = split_host_port(config['listen'], None) + host, port = split_host_port(listen, None) except Exception: raise ValueError('Invalid "restapi" config: expected : for "listen", but got "{0}"' - .format(config['listen'])) + .format(listen)) - if self.__listen is not None: # changing config in runtime + reloading_config = self.__listen is not None # changing config in runtime + if reloading_config: self.shutdown() - self.__listen = config['listen'] - self.__ssl_options = self.__get_ssl_options(config) + self.__listen = listen + self.__ssl_options = ssl_options self.__httpserver_init(host, port) Thread.__init__(self, target=self.serve_forever) self._set_fd_cloexec(self.socket) - self.__protocol = 'http' - # wrap socket with ssl if 'certfile' is defined in a config.yaml # Sometime it's also needed to pass reference to a 'keyfile'. - if self.__ssl_options.get('certfile'): + self.__protocol = 'https' if ssl_options.get('certfile') else 'http' + if self.__protocol == 'https': import ssl - ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) - ctx.load_cert_chain(**self.__ssl_options) + ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH, cafile=ssl_options.get('cafile')) + ctx.load_cert_chain(certfile=ssl_options['certfile'], keyfile=ssl_options.get('keyfile')) + verify_client = ssl_options.get('verify_client') + if verify_client: + modes = {'none': ssl.CERT_NONE, 'optional': ssl.CERT_OPTIONAL, 'required': ssl.CERT_REQUIRED} + if verify_client in modes: + ctx.verify_mode = modes[verify_client] + else: + logger.error('Bad value in the "restapi.verify_client": %s', verify_client) self.socket = ctx.wrap_socket(self.socket, server_side=True) - self.__protocol = 'https' - return True + if reloading_config: + self.start() def reload_config(self, config): if 'listen' not in config: # changing config in runtime raise ValueError('Can not find "restapi.listen" config') - elif (self.__listen != config['listen'] or self.__ssl_options != self.__get_ssl_options(config)) \ - and self.__initialize(config): - self.start() - self.__set_config_parameters(config) + ssl_options = {n: config[n] for n in ('certfile', 'keyfile', 'cafile') if n in config} + + if isinstance(config.get('verify_client'), six.string_types): + ssl_options['verify_client'] = config['verify_client'].lower() + + if self.__listen != config['listen'] or self.__ssl_options != ssl_options: + self.__initialize(config['listen'], ssl_options) + + self.__auth_key = base64.b64encode(config['auth'].encode('utf-8')).decode('utf-8') if 'auth' in config else None + self.connection_string = uri(self.__protocol, config.get('connect_address') or self.__listen, 'patroni') @staticmethod def handle_error(request, client_address): diff --git a/patroni/config.py b/patroni/config.py index ab737f92..ea4a6d7e 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -236,7 +236,8 @@ class Config(object): if value: ret[section][param] = value - _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile']) + _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile', 'cafile', 'verify_client']) + _set_section_values('ctl', ['insecure', 'cacert', 'certfile', 'keyfile']) _set_section_values('postgresql', ['listen', 'connect_address', 'config_dir', 'data_dir', 'pgpass', 'bin_dir']) _set_section_values('log', ['level', 'format', 'dateformat', 'max_queue_size', 'dir', 'file_size', 'file_num', 'loggers']) diff --git a/patroni/ctl.py b/patroni/ctl.py index 8b7cd08f..fe2a6828 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -2,7 +2,6 @@ Patroni Control ''' -import base64 import click import codecs import datetime @@ -15,7 +14,6 @@ import json import logging import os import random -import requests import subprocess import sys import tempfile @@ -30,6 +28,7 @@ from patroni.dcs import get_dcs as _get_dcs from patroni.exceptions import PatroniException from patroni.postgresql import Postgresql from patroni.postgresql.misc import postgres_version_to_int +from patroni.request import PatroniRequest from patroni.utils import patch_config, polling_loop from patroni.version import __version__ from prettytable import PrettyTable @@ -135,36 +134,12 @@ def get_dcs(config, scope): raise PatroniCtlException(str(e)) -def auth_header(config): - if config.get('restapi', {}).get('auth', ''): - return {'Authorization': 'Basic ' + base64.b64encode(config['restapi']['auth'].encode('utf-8')).decode('utf-8')} - - -def request_patroni(member, request_type, endpoint, content=None, headers=None): +def request_patroni(member, method='GET', endpoint=None, data=None): ctx = click.get_current_context() # the current click context - headers = headers or {} - url_parts = urlparse(member.api_url) - logging.debug(url_parts) - if 'Content-Type' not in headers: - headers['Content-Type'] = 'application/json' - - url = '{0}://{1}/{2}'.format(url_parts.scheme, url_parts.netloc, endpoint) - - insecure = ctx.obj.get('ctl', {}).get('insecure', False) - # Get certfile if any from several configuration namespace - cert = ctx.obj.get('ctl', {}).get('cacert') or \ - ctx.obj.get('restapi', {}).get('cacert') or \ - ctx.obj.get('restapi', {}).get('certfile') - # In the case we specificaly disable SSL cert verification we don't want to have the warning - if insecure: - verify = False - elif cert: - verify = cert - else: - verify = True - return getattr(requests, request_type)(url, headers=headers, - data=json.dumps(content) if content else None, timeout=60, - verify=verify) + request_executor = ctx.obj.get('__request_patroni') + if not request_executor: + request_executor = ctx.obj['__request_patroni'] = PatroniRequest(ctx.obj) + return request_executor(member, method, endpoint, data) def print_output(columns, rows=None, alignment=None, fmt='pretty', header=True, delimiter='\t'): @@ -457,9 +432,9 @@ def remove(obj, cluster_name, fmt): def check_response(response, member_name, action_name, silent_success=False): - if response.status_code >= 400: + if response.status >= 400: click.echo('Failed: {0} for member {1}, status code={2}, ({3})'.format( - action_name, member_name, response.status_code, response.text + action_name, member_name, response.status, response.data.decode('utf-8') )) return False elif not silent_success: @@ -493,18 +468,17 @@ def reload(obj, cluster_name, member_names, force, role): members = get_members(cluster, cluster_name, member_names, role, force, 'reload') - content = {} for member in members: - r = request_patroni(member, 'post', 'reload', content, auth_header(obj)) - if r.status_code == 200: + r = request_patroni(member, 'post', 'reload') + if r.status == 200: click.echo('No changes to apply on member {0}'.format(member.name)) - elif r.status_code == 202: + elif r.status == 202: click.echo('Reload request received for member {0} and will be processed within {1} seconds'.format( member.name, cluster.config.data.get('loop_wait')) ) else: click.echo('Failed: reload for member {0}, status code={1}, ({2})'.format( - member.name, r.status_code, r.text) + member.name, r.status, r.data.decode('utf-8')) ) @@ -565,19 +539,19 @@ def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, vers for member in members: if 'schedule' in content: if force and member.data.get('scheduled_restart'): - r = request_patroni(member, 'delete', 'restart', headers=auth_header(obj)) + r = request_patroni(member, 'delete', 'restart') check_response(r, member.name, 'flush scheduled restart', True) - r = request_patroni(member, 'post', 'restart', content, auth_header(obj)) - if r.status_code == 200: + r = request_patroni(member, 'post', 'restart', content) + if r.status == 200: click.echo('Success: restart on member {0}'.format(member.name)) - elif r.status_code == 202: + elif r.status == 202: click.echo('Success: restart scheduled on member {0}'.format(member.name)) - elif r.status_code == 409: + elif r.status == 409: click.echo('Failed: another restart is already scheduled on member {0}'.format(member.name)) else: click.echo('Failed: restart for member {0}, status code={1}, ({2})'.format( - member.name, r.status_code, r.text) + member.name, r.status, r.data.decode('utf-8')) ) @@ -593,8 +567,8 @@ def reinit(obj, cluster_name, member_names, force): for member in members: body = {'force': force} while True: - r = request_patroni(member, 'post', 'reinitialize', body, auth_header(obj)) - if not check_response(r, member.name, 'reinitialize') and r.text.endswith(' already in progress') \ + r = request_patroni(member, 'post', 'reinitialize', body) + if not check_response(r, member.name, 'reinitialize') and r.data.endswith(b' already in progress') \ and not force and click.confirm('Do you want to cancel it and reinitialize anyway?'): body['force'] = True continue @@ -682,19 +656,19 @@ def _do_failover_or_switchover(obj, action, cluster_name, master, candidate, for try: member = cluster.leader.member if cluster.leader else cluster.get_member(candidate, False) - r = request_patroni(member, 'post', action, failover_value, auth_header(obj)) + r = request_patroni(member, 'post', action, failover_value) # probably old patroni, which doesn't support switchover yet - if r.status_code == 501 and action == 'switchover' and 'Server does not support this operation' in r.text: - r = request_patroni(member, 'post', 'failover', failover_value, auth_header(obj)) + if r.status == 501 and action == 'switchover' and b'Server does not support this operation' in r.data: + r = request_patroni(member, 'post', 'failover', failover_value) - if r.status_code in (200, 202): + if r.status in (200, 202): logging.debug(r) cluster = dcs.get_cluster() logging.debug(cluster) - click.echo('{0} {1}'.format(timestamp(), r.text)) + click.echo('{0} {1}'.format(timestamp(), r.data.decode('utf-8'))) else: - click.echo('{0} failed, details: {1}, {2}'.format(action.title(), r.status_code, r.text)) + click.echo('{0} failed, details: {1}, {2}'.format(action.title(), r.status, r.data.decode('utf-8'))) return except Exception: logging.exception(r) @@ -919,7 +893,7 @@ def flush(obj, cluster_name, member_names, force, role, target): for member in members: if target == 'restart': if member.data.get('scheduled_restart'): - r = request_patroni(member, 'delete', 'restart', None, auth_header(obj)) + r = request_patroni(member, 'delete', 'restart') check_response(r, member.name, 'flush scheduled restart') else: click.echo('No scheduled restart for member {0}'.format(member.name)) @@ -956,20 +930,20 @@ def toggle_pause(config, cluster_name, paused, wait): for member in members: try: - r = request_patroni(member, 'patch', 'config', {'pause': paused or None}, auth_header(config)) + r = request_patroni(member, 'patch', 'config', {'pause': paused or None}) except Exception as err: logging.warning(str(err)) logging.warning('Member %s is not accessible', member.name) continue - if r.status_code == 200: + if r.status == 200: if wait: wait_until_pause_is_applied(dcs, paused, cluster) else: click.echo('Success: cluster management is {0}'.format(paused and 'paused' or 'resumed')) else: click.echo('Failed: {0} cluster management status code={1}, ({2})'.format( - paused and 'pause' or 'resume', r.status_code, r.text)) + paused and 'pause' or 'resume', r.status, r.data.decode('utf-8'))) break else: raise PatroniCtlException('Can not find accessible cluster member') @@ -1233,8 +1207,8 @@ def version(obj, cluster_name, member_names): if m.api_url: if not member_names or m.name in member_names: try: - response = request_patroni(m, 'get', 'patroni') - data = response.json() + response = request_patroni(m) + data = json.loads(response.data) version = data.get('patroni', {}).get('version') pg_version = data.get('server_version') pg_version_str = " PostgreSQL {0}".format(format_pg_version(pg_version)) if pg_version else "" diff --git a/patroni/ha.py b/patroni/ha.py index 403f5120..5033f56a 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -3,7 +3,6 @@ import functools import json import logging import psycopg2 -import requests import sys import time import uuid @@ -558,16 +557,15 @@ class Ha(object): args=(self.dcs.loop_wait, on_success, self._leader_access_is_restricted)) return promote_message - @staticmethod - def fetch_node_status(member): + def fetch_node_status(self, member): """This function perform http get request on member.api_url and fetches its status :returns: `_MemberStatus` object """ try: - response = requests.get(member.api_url, timeout=2, verify=False) - logger.info('Got response from %s %s: %s', member.name, member.api_url, response.content) - return _MemberStatus.from_api_response(member, response.json()) + response = self.patroni.request(member, timeout=2, retries=0) + logger.info('Got response from %s %s: %s', member.name, member.api_url, response.data.decode('utf-8')) + return _MemberStatus.from_api_response(member, json.loads(response.data)) except Exception as e: logger.warning("Request failed to %s: GET %s (%s)", member.name, member.api_url, e) return _MemberStatus.unknown(member) diff --git a/patroni/request.py b/patroni/request.py new file mode 100644 index 00000000..3276e2e3 --- /dev/null +++ b/patroni/request.py @@ -0,0 +1,47 @@ +import json +import urllib3 + +from six.moves.urllib_parse import urlparse, urlunparse + + +class PatroniRequest(object): + + def __init__(self, config, insecure=False): + cert_reqs = 'CERT_NONE' if insecure or config.get('ctl', {}).get('insecure', False) else 'CERT_REQUIRED' + self._pool = urllib3.PoolManager(cert_reqs=cert_reqs) + self.reload_config(config) + + @staticmethod + def _get_cfg_value(config, name): + return config.get('ctl', {}).get(name) or config.get('restapi', {}).get(name) + + def _apply_pool_param(self, param, value): + if value: + self._pool.connection_pool_kw[param] = value + else: + self._pool.connection_pool_kw.pop(param, None) + + def _apply_ssl_file_param(self, config, name): + value = self._get_cfg_value(config, name + 'file') + self._apply_pool_param(name + '_file', value) + return value + + def reload_config(self, config): + self._pool.headers = urllib3.make_headers(basic_auth=self._get_cfg_value(config, 'auth')) + + if self._apply_ssl_file_param(config, 'cert'): + self._apply_ssl_file_param(config, 'key') + else: + self._pool.connection_pool_kw.pop('key_file', None) + + cacert = config.get('ctl', {}).get('cacert') or config.get('restapi', {}).get('cafile') + self._apply_pool_param('ca_certs', cacert) + + def __call__(self, member, method='GET', endpoint=None, data=None, **kwargs): + url = member.api_url + if endpoint: + scheme, netloc, _, _, _, _ = urlparse(url) + url = urlunparse((scheme, netloc, endpoint, '', '', '')) + if data is not None: + kwargs['body'] = json.dumps(data) + return self._pool.request(method.upper(), url, **kwargs) diff --git a/tests/test_api.py b/tests/test_api.py index 8aae6e0a..7e4d5b12 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -145,7 +145,7 @@ class MockRestApiServer(RestApiServer): self.serve_forever = Mock() MockRestApiServer._BaseServer__is_shut_down = Mock() MockRestApiServer._BaseServer__shutdown_request = True - config = config or {'listen': '127.0.0.1:8008', 'auth': 'test:test', 'certfile': 'dumb'} + config = config or {'listen': '127.0.0.1:8008', 'auth': 'test:test', 'certfile': 'dumb', 'verify_client': 'a'} super(MockRestApiServer, self).__init__(MockPatroni(), config) Handler(MockRequest(request), ('0.0.0.0', 8080), self) @@ -408,12 +408,18 @@ class TestRestApiServer(unittest.TestCase): def test_reload_config(self): bad_config = {'listen': 'foo'} self.assertRaises(ValueError, MockRestApiServer, None, '', bad_config) - srv = MockRestApiServer(lambda a1, a2, a3: None, '') + srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required'}) self.assertRaises(ValueError, srv.reload_config, bad_config) self.assertRaises(ValueError, srv.reload_config, {}) with patch.object(socket.socket, 'setsockopt', Mock(side_effect=socket.error)): srv.reload_config({'listen': ':8008'}) + def test_check_auth(self): + srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required'}) + mock_rh = Mock() + mock_rh.request.getpeercert.return_value = None + self.assertIsNot(srv.check_auth(mock_rh), True) + def test_handle_error(self): try: raise Exception() diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 2ee90d3c..a190ef3e 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -1,20 +1,21 @@ import etcd +import json import os -import requests import sys import unittest from click.testing import CliRunner from datetime import datetime, timedelta from mock import patch, Mock -from patroni.ctl import ctl, store_config, load_config, output_members, request_patroni, get_dcs, parse_dcs, \ +from patroni.ctl import ctl, store_config, load_config, output_members, get_dcs, parse_dcs, \ get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException, apply_config_changes, \ format_config_for_editing, show_diff, invoke_editor, format_pg_version, find_executable from patroni.dcs.etcd import Client, Failover from patroni.utils import tzutc from psycopg2 import OperationalError +from urllib3 import PoolManager -from . import MockConnect, MockCursor, MockResponse, psycopg2_connect, requests_get +from . import MockConnect, MockCursor, MockResponse, psycopg2_connect from .test_etcd import etcd_read, socket_getaddrinfo from .test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, \ get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader, get_cluster, Member @@ -35,7 +36,7 @@ def test_rw_config(): @patch('patroni.ctl.load_config', Mock(return_value={'scope': 'alpha', 'postgresql': {'data_dir': '.', 'parameters': {}, 'retry_timeout': 5}, - 'restapi': {'auth': 'u:p', 'listen': ''}, 'etcd': {'host': 'localhost:2379'}})) + 'restapi': {'listen': '::', 'certfile': 'a'}, 'etcd': {'host': 'localhost:2379'}})) class TestCtl(unittest.TestCase): @patch('socket.getaddrinfo', socket_getaddrinfo) @@ -75,7 +76,7 @@ class TestCtl(unittest.TestCase): self.assertIsNone(output_members(cluster, name='abc', fmt='tsv')) @patch('patroni.ctl.get_dcs') - @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse())) + @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) def test_switchover(self, mock_get_dcs): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader @@ -125,18 +126,18 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='dummy') assert result.exit_code == 1 - with patch('patroni.ctl.request_patroni', Mock(side_effect=Exception)): + with patch.object(PoolManager, 'request', Mock(side_effect=Exception)): # Non-responding patroni result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n2300-01-01T12:23:00\ny') assert 'falling back to DCS' in result.output - with patch('patroni.ctl.request_patroni') as mocked: - mocked.return_value.status_code = 500 + with patch.object(PoolManager, 'request') as mocked: + mocked.return_value.status = 500 result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny') assert 'Switchover failed' in result.output - mocked.return_value.status_code = 501 - mocked.return_value.text = 'Server does not support this operation' + mocked.return_value.status = 501 + mocked.return_value.data = b'Server does not support this operation' result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny') assert 'Switchover failed' in result.output @@ -151,7 +152,7 @@ class TestCtl(unittest.TestCase): assert result.exit_code == 1 @patch('patroni.ctl.get_dcs') - @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse())) + @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) def test_failover(self, mock_get_dcs): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader @@ -231,7 +232,7 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy']) assert result.exit_code == 1 - @patch('requests.post') + @patch.object(PoolManager, 'request') @patch('patroni.ctl.get_dcs') def test_reload(self, mock_get_dcs, mock_post): mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader @@ -239,18 +240,19 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y') assert 'Failed: reload for member' in result.output - mock_post.return_value.status_code = 200 + mock_post.return_value.status = 200 result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y') assert 'No changes to apply on member' in result.output - mock_post.return_value.status_code = 202 + mock_post.return_value.status = 202 result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y') assert 'Reload request received for member' in result.output - @patch('requests.post', requests_get) + @patch.object(PoolManager, 'request') @patch('patroni.ctl.get_dcs') - def test_restart_reinit(self, mock_get_dcs): + def test_restart_reinit(self, mock_get_dcs, mock_post): mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + mock_post.return_value.status = 503 result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\ny\n') assert 'Failed: restart for' in result.output assert result.exit_code == 0 @@ -285,58 +287,49 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force', '--timeout', '10min']) assert result.exit_code == 0 - with patch('requests.delete', Mock(return_value=MockResponse(500))): - # normal restart, the schedule is actually parsed, but not validated in patronictl - result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', - '--scheduled', '2300-10-01T14:30']) - assert 'Failed: flush scheduled restart' in result.output + # normal restart, the schedule is actually parsed, but not validated in patronictl + result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30']) + assert 'Failed: flush scheduled restart' in result.output with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30']) assert result.exit_code == 1 - with patch('requests.post', Mock(return_value=MockResponse())): - # normal restart, the schedule is actually parsed, but not validated in patronictl - result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0.0', - '--scheduled', '2300-10-01T14:30'], input='y') - assert result.exit_code == 0 - - with patch('requests.post', Mock(return_value=MockResponse(204))): - # get restart with the non-200 return code - # normal restart, the schedule is actually parsed, but not validated in patronictl - result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0', - '--scheduled', '2300-10-01T14:30'], input='y') - assert result.exit_code == 0 - # force restart with restart already present - with patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse(204))): - result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', - '--scheduled', '2300-10-01T14:30']) - assert result.exit_code == 0 + result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30']) + assert result.exit_code == 0 - with patch('requests.post', Mock(return_value=MockResponse(202))): - # get restart with the non-200 return code - # normal restart, the schedule is actually parsed, but not validated in patronictl - result = self.runner.invoke( - ctl, ['restart', 'alpha', '--pg-version', '99.0.0', '--scheduled', '2300-10-01T14:30'], input='y' - ) - assert 'Success: restart scheduled' in result.output - assert result.exit_code == 0 + ctl_args = ['restart', 'alpha', '--pg-version', '99.0', '--scheduled', '2300-10-01T14:30'] + # normal restart, the schedule is actually parsed, but not validated in patronictl + mock_post.return_value.status = 200 + result = self.runner.invoke(ctl, ctl_args, input='y') + assert result.exit_code == 0 - with patch('requests.post', Mock(return_value=MockResponse(409))): - # get restart with the non-200 return code - # normal restart, the schedule is actually parsed, but not validated in patronictl - result = self.runner.invoke( - ctl, ['restart', 'alpha', '--pg-version', '99.0.0', '--scheduled', '2300-10-01T14:30'], input='y' - ) - assert 'Failed: another restart is already' in result.output - assert result.exit_code == 0 + # get restart with the non-200 return code + # normal restart, the schedule is actually parsed, but not validated in patronictl + mock_post.return_value.status = 204 + result = self.runner.invoke(ctl, ctl_args, input='y') + assert result.exit_code == 0 + + # get restart with the non-200 return code + # normal restart, the schedule is actually parsed, but not validated in patronictl + mock_post.return_value.status = 202 + result = self.runner.invoke(ctl, ctl_args, input='y') + assert 'Success: restart scheduled' in result.output + assert result.exit_code == 0 + + # get restart with the non-200 return code + # normal restart, the schedule is actually parsed, but not validated in patronictl + mock_post.return_value.status = 409 + result = self.runner.invoke(ctl, ctl_args, input='y') + assert 'Failed: another restart is already' in result.output + assert result.exit_code == 0 @patch('patroni.ctl.get_dcs') def test_remove(self, mock_get_dcs): mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - result = self.runner.invoke(ctl, ['remove', 'alpha'], input='alpha\nslave') + result = self.runner.invoke(ctl, ['-k', 'remove', 'alpha'], input='alpha\nslave') assert 'Please confirm' in result.output assert 'You are about to remove all' in result.output # Not typing an exact confirmation @@ -353,17 +346,6 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['remove', 'alpha'], input='alpha\nYes I am aware\nleader') assert result.exit_code == 0 - @patch('requests.post', Mock(side_effect=requests.exceptions.ConnectionError('foo'))) - @patch('click.get_current_context') - def test_request_patroni(self, mock_context): - member = get_cluster_initialized_with_leader().leader.member - - mock_context.return_value.obj = {'ctl': {'cacert': 'cert.pem'}} - self.assertRaises(requests.exceptions.ConnectionError, request_patroni, member, 'post', 'dummy', {}) - - mock_context.return_value.obj = {'ctl': {'insecure': True}} - self.assertRaises(requests.exceptions.ConnectionError, request_patroni, member, 'post', 'dummy', {}) - def test_ctl(self): self.runner.invoke(ctl, ['list']) @@ -437,7 +419,7 @@ class TestCtl(unittest.TestCase): assert 'Scheduled restart' in result.output @patch('patroni.ctl.get_dcs') - @patch('requests.delete', Mock(return_value=MockResponse())) + @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) def test_flush(self, mock_get_dcs): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader @@ -447,58 +429,58 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force']) assert 'Success: flush scheduled restart' in result.output - with patch.object(requests, 'delete', return_value=MockResponse(404)): + with patch.object(PoolManager, 'request', return_value=MockResponse(404)): result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force']) assert 'Failed: flush scheduled restart' in result.output + @patch.object(PoolManager, 'request') @patch('patroni.ctl.get_dcs') @patch('patroni.ctl.polling_loop', Mock(return_value=[1])) - def test_pause_cluster(self, mock_get_dcs): + def test_pause_cluster(self, mock_get_dcs, mock_post): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - with patch('requests.patch', Mock(return_value=MockResponse(500))): - result = self.runner.invoke(ctl, ['pause', 'dummy']) - assert 'Failed' in result.output + mock_post.return_value.status = 500 + result = self.runner.invoke(ctl, ['pause', 'dummy']) + assert 'Failed' in result.output - with patch('requests.patch', Mock(return_value=MockResponse(200))),\ - patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): + mock_post.return_value.status = 200 + with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): result = self.runner.invoke(ctl, ['pause', 'dummy']) assert 'Cluster is already paused' in result.output - with patch('requests.patch', Mock(return_value=MockResponse(200))): - result = self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) - assert "'pause' request sent" in result.output - mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), - get_cluster(None, None, [], None, None)]) - self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) - member = Member(1, 'other', 28, {}) - mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), - get_cluster(None, None, [member], None, None)]) - self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + result = self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + assert "'pause' request sent" in result.output + mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), + get_cluster(None, None, [], None, None)]) + self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + member = Member(1, 'other', 28, {}) + mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), + get_cluster(None, None, [member], None, None)]) + self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + @patch.object(PoolManager, 'request') @patch('patroni.ctl.get_dcs') - def test_resume_cluster(self, mock_get_dcs): + def test_resume_cluster(self, mock_get_dcs, mock_post): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + mock_post.return_value.status = 200 + with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=False)): + result = self.runner.invoke(ctl, ['resume', 'dummy']) + assert 'Cluster is not paused' in result.output + with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): - with patch('requests.patch', Mock(return_value=MockResponse(200))): - result = self.runner.invoke(ctl, ['resume', 'dummy']) - assert 'Success' in result.output + result = self.runner.invoke(ctl, ['resume', 'dummy']) + assert 'Success' in result.output - with patch('requests.patch', Mock(return_value=MockResponse(500))): - result = self.runner.invoke(ctl, ['resume', 'dummy']) - assert 'Failed' in result.output + mock_post.return_value.status = 500 + result = self.runner.invoke(ctl, ['resume', 'dummy']) + assert 'Failed' in result.output - with patch('requests.patch', Mock(return_value=MockResponse(200))),\ - patch('patroni.dcs.Cluster.is_paused', Mock(return_value=False)): - result = self.runner.invoke(ctl, ['resume', 'dummy']) - assert 'Cluster is not paused' in result.output - - with patch('requests.patch', Mock(side_effect=Exception)): - result = self.runner.invoke(ctl, ['resume', 'dummy']) - assert 'Can not find accessible cluster member' in result.output + mock_post.side_effect = Exception + result = self.runner.invoke(ctl, ['resume', 'dummy']) + assert 'Can not find accessible cluster member' in result.output def test_apply_config_changes(self): config = {"postgresql": {"parameters": {"work_mem": "4MB"}, "use_pg_rewind": True}, "ttl": 30} @@ -574,13 +556,13 @@ class TestCtl(unittest.TestCase): def test_version(self, mock_get_dcs): mock_get_dcs.return_value = self.e mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - with patch('patroni.ctl.request_patroni') as mocked: + with patch.object(PoolManager, 'request') as mocked: result = self.runner.invoke(ctl, ['version']) assert 'patronictl version' in result.output - mocked.return_value.json = lambda: {'patroni': {'version': '1.2.3'}, 'server_version': 100001} + mocked.return_value.data = json.dumps({'patroni': {'version': '1.2.3'}, 'server_version': 100001}) result = self.runner.invoke(ctl, ['version', 'dummy']) assert '1.2.3' in result.output - with patch('requests.get', Mock(side_effect=Exception)): + with patch.object(PoolManager, 'request', Mock(side_effect=Exception)): result = self.runner.invoke(ctl, ['version', 'dummy']) assert 'failed to get version' in result.output diff --git a/tests/test_ha.py b/tests/test_ha.py index 0ad0c417..4586f6e1 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -137,6 +137,7 @@ zookeeper: self.scheduled_restart = {'schedule': future_restart_time, 'postmaster_start_time': str(postmaster_start_time)} self.watchdog = Watchdog(self.config) + self.request = lambda member, **kwargs: requests_get(member.api_url, **kwargs) def run_async(self, func, args=()): @@ -464,7 +465,6 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'lost leader lock during restart') mock_terminate.assert_called() - @patch('requests.get', requests_get) def test_manual_failover_from_leader(self): self.ha.fetch_node_status = get_node_status() self.ha.has_lock = true @@ -512,7 +512,6 @@ class TestHa(PostgresInit): self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled)) self.assertEqual('no action. i am the leader with the lock', self.ha.run_cycle()) - @patch('requests.get', requests_get) def test_manual_failover_from_leader_in_pause(self): self.ha.has_lock = true self.ha.is_paused = true @@ -522,7 +521,6 @@ class TestHa(PostgresInit): self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None)) self.assertEqual('PAUSE: no action. i am the leader with the lock', self.ha.run_cycle()) - @patch('requests.get', requests_get) def test_manual_failover_from_leader_in_synchronous_mode(self): self.p.is_leader = true self.ha.has_lock = true @@ -535,7 +533,6 @@ class TestHa(PostgresInit): self.ha.is_failover_possible = true self.assertEqual('manual failover: demoting myself', self.ha.run_cycle()) - @patch('requests.get', requests_get) def test_manual_failover_process_no_leader(self): self.p.is_leader = false self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', self.p.name, None)) @@ -585,7 +582,6 @@ class TestHa(PostgresInit): self.ha.is_paused = true self.assertFalse(self.ha.is_healthiest_node()) - @patch('requests.get', requests_get) def test__is_healthiest_node(self): self.ha.cluster = get_cluster_initialized_without_leader(sync=('postgresql1', self.p.name)) self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) @@ -607,7 +603,6 @@ class TestHa(PostgresInit): self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.ha.patroni.nofailover = False - @patch('requests.get', requests_get) def test_fetch_node_status(self): member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'}) self.ha.fetch_node_status(member) @@ -696,7 +691,6 @@ class TestHa(PostgresInit): with patch.object(Leader, 'conn_url', PropertyMock(return_value='')): self.assertEqual(self.ha.run_cycle(), 'continue following the old known standby leader') - @patch('requests.get', requests_get) def test_process_unhealthy_standby_cluster_as_standby_leader(self): self.p.is_leader = false self.p.name = 'leader' @@ -1027,7 +1021,6 @@ class TestHa(PostgresInit): self.assertEqual(self.ha.run_cycle(), 'no action. i am the leader with the lock') @patch('sys.exit', return_value=1) - @patch('requests.get', requests_get) def test_abort_join(self, exit_mock): self.ha.cluster = get_cluster_not_initialized_without_leader() self.p.is_leader = false