URL: https://github.com/freeipa/freeipa/pull/1146
Author: slaykovsky
Title: #1146: rpcserver.py code cleaning
Action: opened
PR body:
"""
Hi!
I was going through `rpcserver` module code and just found out it's very hard
to read. So I've made some style changes and string formatting related changes.
For my point of view, it's now more easy to read.
Any comments welcomed.
Thanks.
"""
To pull the PR as Git branch:
git remote add ghfreeipa https://github.com/freeipa/freeipa
git fetch ghfreeipa pull/1146/head:pr1146
git checkout pr1146
From 470646a36de94d07042d377d4ccebca3fff88185 Mon Sep 17 00:00:00 2001
From: Aleksei Slaikovskii <aslai...@redhat.com>
Date: Wed, 11 Oct 2017 15:40:30 +0200
Subject: [PATCH] rpcserver.py code cleaning
Fixes code formatting, changes old formatting for strings.
---
ipaserver/rpcserver.py | 581 ++++++++++++++++++++++++++++++++-----------------
1 file changed, 381 insertions(+), 200 deletions(-)
diff --git a/ipaserver/rpcserver.py b/ipaserver/rpcserver.py
index 5cbacf406d..ecc3d94fcf 100644
--- a/ipaserver/rpcserver.py
+++ b/ipaserver/rpcserver.py
@@ -1,7 +1,7 @@
# Authors:
# Jason Gerard DeRose <jder...@redhat.com>
#
-# Copyright (C) 2008-2016 Red Hat
+# Copyright (C) 2008-2017 Red Hat
# see file 'COPYING' for use and warranty information
#
# This program is free software; you can redistribute it and/or modify
@@ -45,17 +45,21 @@
from ipalib.frontend import Local
from ipalib.install.kinit import kinit_armor, kinit_password
from ipalib.backend import Executioner
-from ipalib.errors import (PublicError, InternalError, JSONError,
- CCacheError, RefererError, InvalidSessionPassword, NotFound, ACIError,
- ExecutionError, PasswordExpired, KrbPrincipalExpired, UserLocked)
+from ipalib.errors import (
+ PublicError, InternalError, JSONError,
+ CCacheError, RefererError, InvalidSessionPassword,
+ NotFound, ACIError, ExecutionError,
+ PasswordExpired, KrbPrincipalExpired, UserLocked
+)
from ipalib.request import context, destroy_context
-from ipalib.rpc import (xml_dumps, xml_loads,
- json_encode_binary, json_decode_binary)
+from ipalib.rpc import (
+ xml_dumps, xml_loads,
+ json_encode_binary, json_decode_binary
+)
from ipapython.dn import DN
from ipaserver.plugins.ldap2 import ldap2
from ipalib.backend import Backend
-from ipalib.krb_utils import (
- get_credentials_if_valid)
+from ipalib.krb_utils import get_credentials_if_valid
from ipapython import kerberos
from ipapython import ipautil
from ipaplatform.paths import paths
@@ -80,7 +84,7 @@
<body>
<h1>Not Found</h1>
<p>
-The requested URL <strong>%(url)s</strong> was not found on this server.
+The requested URL <strong>{url}</strong> was not found on this server.
</p>
</body>
</html>"""
@@ -92,7 +96,7 @@
<body>
<h1>Bad Request</h1>
<p>
-<strong>%(message)s</strong>
+<strong>{message}</strong>
</p>
</body>
</html>"""
@@ -104,7 +108,7 @@
<body>
<h1>Internal Server Error</h1>
<p>
-<strong>%(message)s</strong>
+<strong>{message}</strong>
</p>
</body>
</html>"""
@@ -116,7 +120,7 @@
<body>
<h1>Invalid Authentication</h1>
<p>
-<strong>%(message)s</strong>
+<strong>{message}</strong>
</p>
</body>
</html>"""
@@ -126,13 +130,14 @@
<title>200 Success</title>
</head>
<body>
-<h1>%(title)s</h1>
+<h1>{title}</h1>
<p>
-<strong>%(message)s</strong>
+<strong>{message}</strong>
</p>
</body>
</html>"""
+
class HTTP_Status(plugable.Plugin):
def not_found(self, environ, start_response, url, message):
"""
@@ -142,8 +147,10 @@ def not_found(self, environ, start_response, url, message):
response_headers = [('Content-Type', 'text/html; charset=utf-8')]
logger.info('%s: URL="%s", %s', status, url, message)
+
start_response(status, response_headers)
- output = _not_found_template % dict(url=escape(url))
+
+ output = _not_found_template.format(url=escape(url))
return [output.encode('utf-8')]
def bad_request(self, environ, start_response, message):
@@ -156,7 +163,8 @@ def bad_request(self, environ, start_response, message):
logger.info('%s: %s', status, message)
start_response(status, response_headers)
- output = _bad_request_template % dict(message=escape(message))
+
+ output = _bad_request_template.format(message=escape(message))
return [output.encode('utf-8')]
def internal_error(self, environ, start_response, message):
@@ -169,7 +177,8 @@ def internal_error(self, environ, start_response, message):
logger.error('%s: %s', status, message)
start_response(status, response_headers)
- output = _internal_error_template % dict(message=escape(message))
+
+ output = _internal_error_template.format(message=escape(message))
return [output.encode('utf-8')]
def unauthorized(self, environ, start_response, message, reason):
@@ -178,15 +187,18 @@ def unauthorized(self, environ, start_response, message, reason):
"""
status = '401 Unauthorized'
response_headers = [('Content-Type', 'text/html; charset=utf-8')]
+
if reason:
response_headers.append(('X-IPA-Rejection-Reason', reason))
logger.info('%s: %s', status, message)
start_response(status, response_headers)
- output = _unauthorized_template % dict(message=escape(message))
+
+ output = _unauthorized_template.format(message=escape(message))
return [output.encode('utf-8')]
+
def read_input(environ):
"""
Read the request body from environ['wsgi.input'].
@@ -200,22 +212,25 @@ def read_input(environ):
def params_2_args_options(params):
if len(params) == 0:
- return (tuple(), dict())
+ return tuple(), dict()
+
if len(params) == 1:
- return (params[0], dict())
- return (params[0], params[1])
+ return params[0], dict()
+
+ return params[0], params[1]
def nicify_query(query, encoding='utf-8'):
if not query:
return
- for (key, value) in query.items():
+
+ for key, value in query.items():
if len(value) == 0:
- yield (key, None)
+ yield key, None
elif len(value) == 1:
- yield (key, value[0].decode(encoding))
+ yield key, value[0].decode(encoding)
else:
- yield (key, tuple(v.decode(encoding) for v in value))
+ yield key, tuple(v.decode(encoding) for v in value)
def extract_query(environ):
@@ -223,16 +238,20 @@ def extract_query(environ):
Return the query as a ``dict``, or ``None`` if no query is presest.
"""
qstr = None
+
if environ['REQUEST_METHOD'] == 'POST':
if environ['CONTENT_TYPE'] == 'application/x-www-form-urlencoded':
qstr = read_input(environ)
elif environ['REQUEST_METHOD'] == 'GET':
qstr = environ['QUERY_STRING']
+
if qstr:
query = dict(nicify_query(parse_qs(qstr))) # keep_blank_values=True)
else:
query = {}
+
environ['wsgi.query'] = query
+
return query
@@ -275,9 +294,13 @@ def route(self, environ, start_response):
if key in self.__apps:
app = self.__apps[key]
return app(environ, start_response)
+
url = environ['SCRIPT_NAME'] + environ['PATH_INFO']
- return self.not_found(environ, start_response, url,
- 'URL fragment "%s" does not have a handler' % (key))
+
+ return self.not_found(
+ environ, start_response, url,
+ 'URL fragment "{}" does not have a handler'.format(key)
+ )
def mount(self, app, key):
"""
@@ -288,16 +311,15 @@ def mount(self, app, key):
# self.name, app, key)
# )
if key in self.__apps:
- raise Exception('%s.mount(): cannot replace %r with %r at %r' % (
- self.name, self.__apps[key], app, key)
+ raise Exception(
+ '{0}.mount(): cannot replace {1} with {2} at {3}'.format(
+ self.name, self.__apps[key], app, key
+ )
)
logger.debug('Mounting %r at %r', app, key)
self.__apps[key] = app
-
-
-
class WSGIExecutioner(Executioner):
"""
Base class for execution backends with a WSGI application interface.
@@ -339,30 +361,42 @@ def wsgi_execute(self, environ):
args = ()
options = {}
command = None
-
e = None
- if not 'HTTP_REFERER' in environ:
+
+ if 'HTTP_REFERER' not in environ:
return self.marshal(result, RefererError(referer='missing'), _id)
- if not environ['HTTP_REFERER'].startswith('https://%s/ipa' % self.api.env.host) and not self.env.in_tree:
- return self.marshal(result, RefererError(referer=environ['HTTP_REFERER']), _id)
+
+ if (
+ not environ['HTTP_REFERER'].startswith(
+ 'https://{}/ipa'.format(self.api.env.host)
+ ) and not self.env.in_tree
+ ):
+ return self.marshal(
+ result, RefererError(referer=environ['HTTP_REFERER']), _id
+ )
+
try:
- if ('HTTP_ACCEPT_LANGUAGE' in environ):
+ if 'HTTP_ACCEPT_LANGUAGE' in environ:
lang_reg_w_q = environ['HTTP_ACCEPT_LANGUAGE'].split(',')[0]
lang_reg = lang_reg_w_q.split(';')[0]
lang_ = lang_reg.split('-')[0]
+
if '-' in lang_reg:
reg = lang_reg.split('-')[1].upper()
else:
reg = lang_.upper()
- os.environ['LANG'] = '%s_%s' % (lang_, reg)
+
+ os.environ['LANG'] = '{0}_{1}'.format(lang_, reg)
+
if (
environ.get('CONTENT_TYPE', '').startswith(self.content_type)
and environ['REQUEST_METHOD'] == 'POST'
):
data = read_input(environ)
- (name, args, options, _id) = self.unmarshal(data)
+ name, args, options, _id = self.unmarshal(data)
else:
- (name, args, options, _id) = self.simple_unmarshal(environ)
+ name, args, options, _id = self.simple_unmarshal(environ)
+
if name in self._system_commands:
result = self._system_commands[name](self, *args, **options)
else:
@@ -370,8 +404,10 @@ def wsgi_execute(self, environ):
result = command(*args, **options)
except PublicError as e:
if self.api.env.debug:
- logger.debug('WSGI wsgi_execute PublicError: %s',
- traceback.format_exc())
+ logger.debug(
+ 'WSGI wsgi_execute PublicError: %s',
+ traceback.format_exc()
+ )
error = e
except Exception as e:
logger.exception(
@@ -382,6 +418,7 @@ def wsgi_execute(self, environ):
os.environ['LANG'] = lang
principal = getattr(context, 'principal', 'UNKNOWN')
+
if command is not None:
try:
params = command.args_options_2_params(*args, **options)
@@ -393,24 +430,25 @@ def wsgi_execute(self, environ):
# get at least some context of what is going on
params = options
error = e
+
if error:
result_string = type(error).__name__
else:
result_string = 'SUCCESS'
- logger.info('[%s] %s: %s(%s): %s',
- type(self).__name__,
- principal,
- name,
- ', '.join(command._repr_iter(**params)),
- result_string)
+
+ logger.info(
+ '[%s] %s: %s(%s): %s',
+ type(self).__name__, principal, name,
+ ', '.join(command._repr_iter(**params)), result_string
+ )
else:
- logger.info('[%s] %s: %s: %s',
- type(self).__name__,
- principal,
- name,
- type(error).__name__)
+ logger.info(
+ '[%s] %s: %s: %s',
+ type(self).__name__, principal, name, type(error).__name__
+ )
version = options.get('version', VERSION_WITHOUT_CAPABILITIES)
+
return self.marshal(result, error, _id, version)
def simple_unmarshal(self, environ):
@@ -427,11 +465,14 @@ def __call__(self, environ, start_response):
try:
status = HTTP_STATUS_SUCCESS
response = self.wsgi_execute(environ)
+
if self.headers:
headers = self.headers
else:
- headers = [('Content-Type',
- self.content_type + '; charset=utf-8')]
+ headers = [(
+ 'Content-Type',
+ '{}; charset=utf-8'.format(self.content_type)
+ )]
except Exception:
logger.exception('WSGI %s.__call__():', self.name)
status = HTTP_STATUS_SERVER_ERROR
@@ -443,14 +484,17 @@ def __call__(self, environ, start_response):
headers.append(('IPASESSION', logout_cookie))
start_response(status, headers)
+
return [response]
def unmarshal(self, data):
- raise NotImplementedError('%s.unmarshal()' % type(self).__name__)
+ raise NotImplementedError('{}.unmarshal()'.format(type(self).__name__))
- def marshal(self, result, error, _id=None,
- version=VERSION_WITHOUT_CAPABILITIES):
- raise NotImplementedError('%s.marshal()' % type(self).__name__)
+ def marshal(
+ self, result, error, _id=None,
+ version=VERSION_WITHOUT_CAPABILITIES
+ ):
+ raise NotImplementedError('{}.marshal()'.format(type(self).__name__))
class jsonserver(WSGIExecutioner, HTTP_Status):
@@ -465,35 +509,31 @@ class jsonserver(WSGIExecutioner, HTTP_Status):
content_type = 'application/json'
def __call__(self, environ, start_response):
- '''
- '''
-
logger.debug('WSGI jsonserver.__call__:')
response = super(jsonserver, self).__call__(environ, start_response)
return response
- def marshal(self, result, error, _id=None,
- version=VERSION_WITHOUT_CAPABILITIES):
+ def marshal(
+ self, result, error, _id=None,
+ version=VERSION_WITHOUT_CAPABILITIES
+ ):
if error:
assert isinstance(error, PublicError)
error = dict(
- code=error.errno,
- message=error.strerror,
- data=error.kw,
- name=unicode(error.__class__.__name__),
+ code=error.errno, message=error.strerror,
+ data=error.kw, name=unicode(error.__class__.__name__),
)
+
principal = getattr(context, 'principal', 'UNKNOWN')
response = dict(
- result=result,
- error=error,
- id=_id,
- principal=unicode(principal),
- version=unicode(VERSION),
+ result=result, error=error, id=_id,
+ principal=unicode(principal), version=unicode(VERSION),
)
dump = json_encode_binary(
response, version, pretty_print=self.api.env.debug
)
+
return dump.encode('utf-8')
def unmarshal(self, data):
@@ -501,26 +541,36 @@ def unmarshal(self, data):
d = json_decode_binary(data)
except ValueError as e:
raise JSONError(error=e)
+
if not isinstance(d, dict):
raise JSONError(error=_('Request must be a dict'))
+
if 'method' not in d:
raise JSONError(error=_('Request is missing "method"'))
+
if 'params' not in d:
raise JSONError(error=_('Request is missing "params"'))
+
method = d['method']
params = d['params']
_id = d.get('id')
+
if not isinstance(params, (list, tuple)):
raise JSONError(error=_('params must be a list'))
+
if len(params) != 2:
raise JSONError(error=_('params must contain [args, options]'))
+
args = params[0]
if not isinstance(args, (list, tuple)):
raise JSONError(error=_('params[0] (aka args) must be a list'))
+
options = params[1]
if not isinstance(options, dict):
raise JSONError(error=_('params[1] (aka options) must be a dict'))
+
options = dict((str(k), v) for (k, v) in options.items())
+
return (method, args, options, _id)
@@ -534,6 +584,7 @@ def __init__(self, target_host, ccache_name=None):
def __call__(self, request):
self.initial_step(request)
request.register_hook('response', self.handle_response)
+
return request
def deregister(self, response):
@@ -543,24 +594,31 @@ def _get_negotiate_token(self, response):
token = None
if response is not None:
h = response.headers.get('www-authenticate', '')
+
if h.startswith('Negotiate'):
val = h[h.find('Negotiate') + len('Negotiate'):].strip()
+
if len(val) > 0:
token = b64decode(val)
+
return token
def _set_authz_header(self, request, token):
request.headers['Authorization'] = (
- 'Negotiate {}'.format(b64encode(token).decode('utf-8')))
+ 'Negotiate {}'.format(b64encode(token).decode('utf-8'))
+ )
def initial_step(self, request, response=None):
if self.context is None:
store = {'ccache': self.ccache_name}
creds = gssapi.Credentials(usage='initiate', store=store)
- name = gssapi.Name('HTTP@{0}'.format(self.target_host),
- name_type=gssapi.NameType.hostbased_service)
- self.context = gssapi.SecurityContext(creds=creds, name=name,
- usage='initiate')
+ name = gssapi.Name(
+ 'HTTP@{0}'.format(self.target_host),
+ name_type=gssapi.NameType.hostbased_service
+ )
+ self.context = gssapi.SecurityContext(
+ creds=creds, name=name, usage='initiate'
+ )
in_token = self._get_negotiate_token(response)
out_token = self.context.step(in_token)
@@ -574,6 +632,7 @@ def handle_response(self, response, **kwargs):
in_token = self._get_negotiate_token(response)
if in_token is not None:
out_token = self.context.step(in_token)
+
if self.context.complete:
return response
elif not out_token:
@@ -625,8 +684,9 @@ def get_environ_creds(self, environ):
gss_name = gssapi.Name(principal, gssapi.NameType.kerberos_principal)
# Fail if Kerberos credentials are expired or missing
- creds = get_credentials_if_valid(name=gss_name,
- ccache_name=ccache_name)
+ creds = get_credentials_if_valid(
+ name=gss_name, ccache_name=ccache_name
+ )
if not creds:
logger.debug(
'ccache expired or invalid, deleting session, need login')
@@ -634,8 +694,10 @@ def get_environ_creds(self, environ):
return ccache_name
-
- def finalize_kerberos_acquisition(self, who, ccache_name, environ, start_response, headers=None):
+ def finalize_kerberos_acquisition(
+ self, who, ccache_name,
+ environ, start_response, headers=None
+ ):
if headers is None:
headers = []
@@ -643,16 +705,21 @@ def finalize_kerberos_acquisition(self, who, ccache_name, environ, start_respons
# generate a cookie for us.
try:
target = self.api.env.host
- r = requests.get('http://{0}/ipa/session/cookie'.format(target),
- auth=NegotiateAuth(target, ccache_name),
- verify=paths.IPA_CA_CRT)
+ r = requests.get(
+ 'http://{0}/ipa/session/cookie'.format(target),
+ auth=NegotiateAuth(target, ccache_name),
+ verify=paths.IPA_CA_CRT
+ )
session_cookie = r.cookies.get("ipa_session")
+
if not session_cookie:
raise ValueError('No session cookie found')
+
except Exception as e:
- return self.unauthorized(environ, start_response,
- str(e),
- 'Authentication failed')
+ return self.unauthorized(
+ environ, start_response,
+ str(e), 'Authentication failed'
+ )
headers.append(('IPASESSION', session_cookie))
@@ -669,11 +736,11 @@ def _on_finalize(self):
def __call__(self, environ, start_response):
logger.debug('KerberosWSGIExecutioner.__call__:')
- user_ccache=environ.get('KRB5CCNAME')
+ user_ccache = environ.get('KRB5CCNAME')
object.__setattr__(
self, 'headers',
- [('Content-Type', '%s; charset=utf-8' % self.content_type)]
+ [('Content-Type', '{}; charset=utf-8'.format(self.content_type))]
)
if user_ccache is None:
@@ -683,14 +750,16 @@ def __call__(self, environ, start_response):
logger.error(
'%s: %s', status,
'KerberosWSGIExecutioner.__call__: '
- 'KRB5CCNAME not defined in HTTP request environment')
+ 'KRB5CCNAME not defined in HTTP request environment'
+ )
return self.marshal(None, CCacheError())
try:
self.create_context(ccache=user_ccache)
response = super(KerberosWSGIExecutioner, self).__call__(
- environ, start_response)
+ environ, start_response
+ )
except PublicError as e:
status = HTTP_STATUS_SUCCESS
response = status.encode('utf-8')
@@ -698,6 +767,7 @@ def __call__(self, environ, start_response):
return self.marshal(None, e)
finally:
destroy_context()
+
return response
@@ -715,8 +785,11 @@ def listMethods(self, *params):
"""list methods for XML-RPC introspection"""
if params:
raise errors.ZeroArgumentError(name='system.listMethods')
- return (tuple(unicode(cmd.name) for cmd in self.api.Command) +
- tuple(unicode(name) for name in self._system_commands))
+
+ return (
+ tuple(unicode(cmd.name) for cmd in self.api.Command) +
+ tuple(unicode(name) for name in self._system_commands)
+ )
def _get_method_name(self, name, *params):
"""Get a method name for XML-RPC introspection commands"""
@@ -724,12 +797,15 @@ def _get_method_name(self, name, *params):
raise errors.RequirementError(name='method name')
elif len(params) > 1:
raise errors.MaxArgumentError(name=name, count=1)
+
[method_name] = params
+
return method_name
def methodSignature(self, *params):
"""get method signature for XML-RPC introspection"""
method_name = self._get_method_name('system.methodSignature', *params)
+
if method_name in self._system_commands:
# TODO
# for now let's not go out of our way to document standard XML-RPC
@@ -744,10 +820,12 @@ def methodSignature(self, *params):
def methodHelp(self, *params):
"""get method docstring for XML-RPC introspection"""
method_name = self._get_method_name('system.methodHelp', *params)
+
if method_name in self._system_commands:
return u''
else:
command = self._get_command(method_name)
+
return unicode(command.doc or '')
_system_commands = {
@@ -757,29 +835,40 @@ def methodHelp(self, *params):
}
def unmarshal(self, data):
- (params, name) = xml_loads(data)
+ params, name = xml_loads(data)
+
if name in self._system_commands:
# For XML-RPC introspection, return params directly
- return (name, params, {}, None)
- (args, options) = params_2_args_options(params)
+ return name, params, {}, None
+
+ args, options = params_2_args_options(params)
+
if 'version' not in options:
# Keep backwards compatibility with client containing
# bug https://fedorahosted.org/freeipa/ticket/3294:
# If `version` is not given in XML-RPC, assume an old version
options['version'] = VERSION_WITHOUT_CAPABILITIES
- return (name, args, options, None)
- def marshal(self, result, error, _id=None,
- version=VERSION_WITHOUT_CAPABILITIES):
+ return name, args, options, None
+
+ def marshal(
+ self, result, error, _id=None,
+ version=VERSION_WITHOUT_CAPABILITIES
+ ):
if error:
- logger.debug('response: %s: %s',
- error.__class__.__name__, str(error))
+ logger.debug(
+ 'response: %s: %s',
+ error.__class__.__name__, str(error)
+ )
response = Fault(error.errno, error.strerror)
else:
if isinstance(result, dict):
- logger.debug('response: entries returned %d',
- result.get('count', 1))
+ logger.debug(
+ 'response: entries returned %d',
+ result.get('count', 1)
+ )
response = (result,)
+
dump = xml_dumps(response, version, methodresponse=True)
return dump.encode('utf-8')
@@ -818,7 +907,9 @@ def __call__(self, environ, start_response):
return self.unauthorized(environ, start_response, str(e), 'denied')
try:
- response = super(jsonserver_session, self).__call__(environ, start_response)
+ response = super(jsonserver_session, self).__call__(
+ environ, start_response
+ )
finally:
destroy_context()
@@ -848,7 +939,9 @@ def __call__(self, environ, start_response):
if user_ccache_name is None:
return self.need_login(start_response)
- return self.finalize_kerberos_acquisition('login_kerberos', user_ccache_name, environ, start_response)
+ return self.finalize_kerberos_acquisition(
+ 'login_kerberos', user_ccache_name, environ, start_response
+ )
class login_kerberos(KerberosLogin):
@@ -863,8 +956,9 @@ def __call__(self, environ, start_response):
if 'KRB5CCNAME' not in environ:
return self.unauthorized(
- environ, start_response, 'KRB5CCNAME not set',
- 'Authentication failed')
+ environ, start_response,
+ 'KRB5CCNAME not set', 'Authentication failed'
+ )
return super(login_x509, self).__call__(environ, start_response)
@@ -884,27 +978,38 @@ def __call__(self, environ, start_response):
# Get the user and password parameters from the request
content_type = environ.get('CONTENT_TYPE', '').lower()
if not content_type.startswith('application/x-www-form-urlencoded'):
- return self.bad_request(environ, start_response, "Content-Type must be application/x-www-form-urlencoded")
+ return self.bad_request(
+ environ, start_response,
+ "Content-Type must be application/x-www-form-urlencoded"
+ )
method = environ.get('REQUEST_METHOD', '').upper()
if method == 'POST':
query_string = read_input(environ)
else:
- return self.bad_request(environ, start_response, "HTTP request method must be POST")
+ return self.bad_request(
+ environ, start_response, "HTTP request method must be POST"
+ )
try:
query_dict = parse_qs(query_string)
except Exception as e:
- return self.bad_request(environ, start_response, "cannot parse query data")
+ return self.bad_request(
+ environ, start_response, "cannot parse query data"
+ )
user = query_dict.get('user', None)
if user is not None:
if len(user) == 1:
user = user[0]
else:
- return self.bad_request(environ, start_response, "more than one user parameter")
+ return self.bad_request(
+ environ, start_response, "more than one user parameter"
+ )
else:
- return self.bad_request(environ, start_response, "no user specified")
+ return self.bad_request(
+ environ, start_response, "no user specified"
+ )
# allows login in the form user@SERVER_REALM or user@server_realm
# we kinit as enterprise principal so we can assume that unknown realms
@@ -921,49 +1026,62 @@ def __call__(self, environ, start_response):
if len(password) == 1:
password = password[0]
else:
- return self.bad_request(environ, start_response, "more than one password parameter")
+ return self.bad_request(
+ environ, start_response, "more than one password parameter"
+ )
else:
- return self.bad_request(environ, start_response, "no password specified")
+ return self.bad_request(
+ environ, start_response, "no password specified"
+ )
+
+ # Get the ccache we'll use and attempt to get credentials in it with
+ # user,password
+ ipa_ccache_name = os.path.join(
+ paths.IPA_CCACHES, 'kinit_{}'.format(os.getpid())
+ )
- # Get the ccache we'll use and attempt to get credentials in it with user,password
- ipa_ccache_name = os.path.join(paths.IPA_CCACHES,
- 'kinit_{}'.format(os.getpid()))
try:
# try to remove in case an old file was there
os.unlink(ipa_ccache_name)
except OSError:
pass
+
try:
self.kinit(user_principal, password, ipa_ccache_name)
except PasswordExpired as e:
- return self.unauthorized(environ, start_response, str(e), 'password-expired')
+ return self.unauthorized(
+ environ, start_response, str(e), 'password-expired'
+ )
except InvalidSessionPassword as e:
- return self.unauthorized(environ, start_response, str(e), 'invalid-password')
+ return self.unauthorized(
+ environ, start_response, str(e), 'invalid-password'
+ )
except KrbPrincipalExpired as e:
- return self.unauthorized(environ,
- start_response,
- str(e),
- 'krbprincipal-expired')
+ return self.unauthorized(
+ environ, start_response, str(e), 'krbprincipal-expired'
+ )
except UserLocked as e:
- return self.unauthorized(environ,
- start_response,
- str(e),
- 'user-locked')
-
- result = self.finalize_kerberos_acquisition('login_password',
- ipa_ccache_name, environ,
- start_response)
+ return self.unauthorized(
+ environ, start_response, str(e), 'user-locked'
+ )
+
+ result = self.finalize_kerberos_acquisition(
+ 'login_password', ipa_ccache_name, environ, start_response
+ )
+
try:
# Try not to litter the filesystem with unused TGTs
os.unlink(ipa_ccache_name)
except OSError:
pass
+
return result
def kinit(self, principal, password, ccache_name):
# get anonymous ccache as an armor for FAST to enable OTP auth
- armor_path = os.path.join(paths.IPA_CCACHES,
- "armor_{}".format(os.getpid()))
+ armor_path = os.path.join(
+ paths.IPA_CCACHES, "armor_{}".format(os.getpid())
+ )
logger.debug('Obtaining armor in ccache %s', armor_path)
@@ -979,31 +1097,40 @@ def kinit(self, principal, password, ccache_name):
try:
kinit_password(
- unicode(principal),
- password,
- ccache_name,
- armor_ccache_name=armor_path,
- enterprise=True,
- lifetime=self.api.env.kinit_lifetime)
+ unicode(principal), password,
+ ccache_name, armor_ccache_name=armor_path,
+ enterprise=True, lifetime=self.api.env.kinit_lifetime
+ )
if armor_path:
logger.debug('Cleanup the armor ccache')
- ipautil.run([paths.KDESTROY, '-A', '-c', armor_path],
- env={'KRB5CCNAME': armor_path}, raiseonerr=False)
+ ipautil.run(
+ [paths.KDESTROY, '-A', '-c', armor_path],
+ env={'KRB5CCNAME': armor_path}, raiseonerr=False
+ )
except RuntimeError as e:
- if ('kinit: Cannot read password while '
- 'getting initial credentials') in str(e):
+ if (
+ 'kinit: Cannot read password while '
+ 'getting initial credentials'
+ ) in str(e):
raise PasswordExpired(principal=principal, message=unicode(e))
- elif ('kinit: Client\'s entry in database'
- ' has expired while getting initial credentials') in str(e):
- raise KrbPrincipalExpired(principal=principal,
- message=unicode(e))
- elif ('kinit: Clients credentials have been revoked '
- 'while getting initial credentials') in str(e):
- raise UserLocked(principal=principal,
- message=unicode(e))
- raise InvalidSessionPassword(principal=principal,
- message=unicode(e))
+ elif (
+ 'kinit: Client\'s entry in database'
+ ' has expired while getting initial credentials'
+ ) in str(e):
+ raise KrbPrincipalExpired(
+ principal=principal, message=unicode(e)
+ )
+ elif (
+ 'kinit: Clients credentials have been revoked '
+ 'while getting initial credentials'
+ ) in str(e):
+ raise UserLocked(
+ principal=principal, message=unicode(e)
+ )
+ raise InvalidSessionPassword(
+ principal=principal, message=unicode(e)
+ )
class change_password(Backend, HTTP_Status):
@@ -1021,18 +1148,25 @@ def __call__(self, environ, start_response):
# Get the user and password parameters from the request
content_type = environ.get('CONTENT_TYPE', '').lower()
if not content_type.startswith('application/x-www-form-urlencoded'):
- return self.bad_request(environ, start_response, "Content-Type must be application/x-www-form-urlencoded")
+ return self.bad_request(
+ environ, start_response,
+ "Content-Type must be application/x-www-form-urlencoded"
+ )
method = environ.get('REQUEST_METHOD', '').upper()
if method == 'POST':
query_string = read_input(environ)
else:
- return self.bad_request(environ, start_response, "HTTP request method must be POST")
+ return self.bad_request(
+ environ, start_response, "HTTP request method must be POST"
+ )
try:
query_dict = parse_qs(query_string)
except Exception as e:
- return self.bad_request(environ, start_response, "cannot parse query data")
+ return self.bad_request(
+ environ, start_response, "cannot parse query data"
+ )
data = {}
for field in ('user', 'old_password', 'new_password', 'otp'):
@@ -1041,27 +1175,37 @@ def __call__(self, environ, start_response):
if len(value) == 1:
data[field] = value[0]
else:
- return self.bad_request(environ, start_response, "more than one %s parameter"
- % field)
+ return self.bad_request(
+ environ, start_response,
+ "more than one {} parameter".format(field)
+ )
elif field != 'otp': # otp is optional
- return self.bad_request(environ, start_response, "no %s specified" % field)
+ return self.bad_request(
+ environ, start_response, "no {} specified".format(field)
+ )
# start building the response
- logger.info("WSGI change_password: start password change of user '%s'",
- data['user'])
+ logger.info(
+ "WSGI change_password: start password change of user '%s'",
+ data['user']
+ )
+
status = HTTP_STATUS_SUCCESS
response_headers = [('Content-Type', 'text/html; charset=utf-8')]
title = 'Password change rejected'
result = 'error'
policy_error = None
- bind_dn = DN((self.api.Object.user.primary_key.name, data['user']),
- self.api.env.container_user, self.api.env.basedn)
+ bind_dn = DN(
+ (self.api.Object.user.primary_key.name, data['user']),
+ self.api.env.container_user, self.api.env.basedn
+ )
try:
pw = data['old_password']
if data.get('otp'):
pw = data['old_password'] + data['otp']
+
conn = ldap2(self.api)
conn.connect(bind_dn=bind_dn, bind_pw=pw)
except (NotFound, ACIError):
@@ -1069,21 +1213,28 @@ def __call__(self, environ, start_response):
message = 'The old password or username is not correct.'
except Exception as e:
message = "Could not connect to LDAP server."
- logger.error("change_password: cannot authenticate '%s' to LDAP "
- "server: %s",
- data['user'], str(e))
+ logger.error(
+ "change_password: cannot authenticate '%s' to LDAP server: %s",
+ data['user'], str(e)
+ )
else:
try:
- conn.modify_password(bind_dn, data['new_password'], data['old_password'], skip_bind=True)
+ conn.modify_password(
+ bind_dn, data['new_password'],
+ data['old_password'], skip_bind=True
+ )
except ExecutionError as e:
result = 'policy-error'
policy_error = escape(str(e))
- message = "Password change was rejected: %s" % escape(str(e))
+ message = "Password change was rejected: {}".format(
+ escape(str(e))
+ )
except Exception as e:
message = "Could not change the password"
- logger.error("change_password: cannot change password of "
- "'%s': %s",
- data['user'], str(e))
+ logger.error(
+ "change_password: cannot change password of '%s': %s",
+ data['user'], str(e)
+ )
else:
result = 'ok'
title = "Password change successful"
@@ -1096,13 +1247,19 @@ def __call__(self, environ, start_response):
response_headers.append(('X-IPA-Pwchange-Result', result))
if policy_error:
- response_headers.append(('X-IPA-Pwchange-Policy-Error', policy_error))
+ response_headers.append(
+ ('X-IPA-Pwchange-Policy-Error', policy_error)
+ )
start_response(status, response_headers)
- output = _success_template % dict(title=str(title),
- message=str(message))
+
+ output = _success_template % dict(
+ title=str(title),
+ message=str(message)
+ )
return [output.encode('utf-8')]
+
class sync_token(Backend, HTTP_Status):
content_type = 'text/plain'
key = '/session/sync_token'
@@ -1124,49 +1281,68 @@ def __call__(self, environ, start_response):
# Make sure this is a form request.
content_type = environ.get('CONTENT_TYPE', '').lower()
if not content_type.startswith('application/x-www-form-urlencoded'):
- return self.bad_request(environ, start_response, "Content-Type must be application/x-www-form-urlencoded")
+ return self.bad_request(
+ environ, start_response,
+ "Content-Type must be application/x-www-form-urlencoded"
+ )
# Make sure this is a POST request.
method = environ.get('REQUEST_METHOD', '').upper()
if method == 'POST':
query_string = read_input(environ)
else:
- return self.bad_request(environ, start_response, "HTTP request method must be POST")
+ return self.bad_request(
+ environ, start_response, "HTTP request method must be POST"
+ )
# Parse the query string to a dictionary.
try:
query_dict = parse_qs(query_string)
except Exception as e:
- return self.bad_request(environ, start_response, "cannot parse query data")
+ return self.bad_request(
+ environ, start_response, "cannot parse query data"
+ )
data = {}
- for field in ('user', 'password', 'first_code', 'second_code', 'token'):
+ for field in (
+ 'user', 'password', 'first_code', 'second_code', 'token'
+ ):
value = query_dict.get(field, None)
if value is not None:
if len(value) == 1:
data[field] = value[0]
else:
- return self.bad_request(environ, start_response, "more than one %s parameter"
- % field)
+ return self.bad_request(
+ environ, start_response,
+ "more than one {} parameter".format(field)
+ )
elif field != 'token':
- return self.bad_request(environ, start_response, "no %s specified" % field)
+ return self.bad_request(
+ environ, start_response, "no {} specified".format(field)
+ )
# Create the request control.
sr = self.OTPSyncRequest()
sr.setComponentByName('firstCode', data['first_code'])
sr.setComponentByName('secondCode', data['second_code'])
+
if 'token' in data:
try:
token_dn = DN(data['token'])
except ValueError:
- token_dn = DN((self.api.Object.otptoken.primary_key.name, data['token']),
- self.api.env.container_otp, self.api.env.basedn)
+ token_dn = DN(
+ (self.api.Object.otptoken.primary_key.name, data['token']),
+ self.api.env.container_otp, self.api.env.basedn
+ )
sr.setComponentByName('tokenDN', str(token_dn))
+
rc = ldap.controls.RequestControl(sr.OID, True, encoder.encode(sr))
# Resolve the user DN
- bind_dn = DN((self.api.Object.user.primary_key.name, data['user']),
- self.api.env.container_user, self.api.env.basedn)
+ bind_dn = DN(
+ (self.api.Object.user.primary_key.name, data['user']),
+ self.api.env.container_user, self.api.env.basedn
+ )
# Start building the response.
status = HTTP_STATUS_SUCCESS
@@ -1176,9 +1352,9 @@ def __call__(self, environ, start_response):
# Perform the synchronization.
conn = ldap2(self.api)
try:
- conn.connect(bind_dn=bind_dn,
- bind_pw=data['password'],
- serverctrls=[rc,])
+ conn.connect(
+ bind_dn=bind_dn, bind_pw=data['password'], serverctrls=[rc, ]
+ )
result = 'ok'
title = "Token sync successful"
message = "Token was synchronized."
@@ -1188,9 +1364,10 @@ def __call__(self, environ, start_response):
except Exception as e:
result = 'error'
message = "Could not connect to LDAP server."
- logger.error("token_sync: cannot authenticate '%s' to LDAP "
- "server: %s",
- data['user'], str(e))
+ logger.error(
+ "token_sync: cannot authenticate '%s' to LDAP server: %s",
+ data['user'], str(e)
+ )
finally:
if conn.isconnected():
conn.disconnect()
@@ -1198,10 +1375,13 @@ def __call__(self, environ, start_response):
# Report status and return.
response_headers.append(('X-IPA-TokenSync-Result', result))
start_response(status, response_headers)
- output = _success_template % dict(title=str(title),
- message=str(message))
+
+ output = _success_template.format(
+ title=str(title), message=str(message)
+ )
return [output.encode('utf-8')]
+
class xmlserver_session(xmlserver, KerberosSession):
"""
XML RPC server protected with session auth.
@@ -1226,9 +1406,6 @@ def need_login(self, start_response):
return [response]
def __call__(self, environ, start_response):
- '''
- '''
-
logger.debug('WSGI xmlserver_session.__call__:')
ccache_name = environ.get('KRB5CCNAME')
@@ -1241,8 +1418,10 @@ def __call__(self, environ, start_response):
# Redirect to /ipa/xml if Kerberos credentials are expired
creds = get_credentials_if_valid(ccache_name=ccache_name)
if not creds:
- logger.debug('xmlserver_session.__call_: ccache expired, deleting '
- 'session, need login')
+ logger.debug(
+ 'xmlserver_session.__call_: ccache expired, deleting '
+ 'session, need login'
+ )
# The request is finished with the ccache, destroy it.
return self.need_login(start_response)
@@ -1250,7 +1429,9 @@ def __call__(self, environ, start_response):
setattr(context, 'ccache_name', ccache_name)
try:
- response = super(xmlserver_session, self).__call__(environ, start_response)
+ response = super(xmlserver_session, self).__call__(
+ environ, start_response
+ )
finally:
destroy_context()
_______________________________________________
FreeIPA-devel mailing list -- freeipa-devel@lists.fedorahosted.org
To unsubscribe send an email to freeipa-devel-le...@lists.fedorahosted.org
