kallithea Changeset - db9718bde286

Changeset - db9718bde286

Parent rev.

Child rev.

[Not reviewed]

stable

0 1 0

Mads Kiilerich - 6 years ago 2020-05-09 02:49:13
mads@kiilerich.com

Grafted from: 15cdbf5e2956

auth: simplify Crowd - drop unused and apparently slightly buggy _request functionality

1 file changed with 1 insertions and 17 deletions:

kallithea/lib/auth_modules/auth_crowd.py

0 comments (0 inline, 0 general)

kallithea/lib/auth_modules/auth_crowd.py

➞

Show inline comments

@@ @@ -57,84 +57,68 @@ class CrowdServer(object): @@
         """
         if "port" not in kwargs:
             kwargs["port"] = "8095"
         self._logger = kwargs.get("logger", logging.getLogger(__name__))
         self._uri = "%s://%s:%s/crowd" % (kwargs.get("method", "http"),
                                     kwargs.get("host", "127.0.0.1"),
                                     kwargs.get("port", "8095"))
         self.set_credentials(kwargs.get("user", ""),
                              kwargs.get("passwd", ""))
         self._version = kwargs.get("version", "latest")
         self._url_list = None
         self._appname = "crowd"
     def set_credentials(self, user, passwd):
         self.user = user
         self.passwd = passwd
         self._make_opener()
     def _make_opener(self):
         mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
         mgr.add_password(None, self._uri, self.user, self.passwd)
         handler = urllib.request.HTTPBasicAuthHandler(mgr)
         self.opener = urllib.request.build_opener(handler)
     def _request(self, url, body=None, headers=None,
                  method=None, noformat=False,
                  empty_response_ok=False):
     def _request(self, url, body=None):
         _headers = {"Content-type": "application/json",
                     "Accept": "application/json"}
         if self.user and self.passwd:
             authstring = ascii_str(base64.b64encode(safe_bytes("%s:%s" % (self.user, self.passwd))))
             _headers["Authorization"] = "Basic %s" % authstring
         if headers:
             _headers.update(headers)
         log.debug("Sent to crowd at %s:\nHeaders: %s\nBody:\n%s", url, _headers, body)
         req = urllib.request.Request(url, body, _headers)
         if method:
             req.get_method = lambda: method
         global msg
         msg = None
         try:
             rdoc = self.opener.open(req)
             msg = rdoc.read()
             if not msg and empty_response_ok:
                 rval = {}
                 rval["status"] = True
                 rval["error"] = "Response body was empty"
             elif not noformat:
                 rval = ext_json.loads(msg)
                 rval["status"] = True
             else:
                 rval = "".join(rdoc.readlines())
         except Exception as e:
             if not noformat:
                 rval = {"status": False,
                         "body": body,
                         "error": "%s\n%r" % (e, msg)}
             else:
                 rval = None
         return rval
     def user_auth(self, username, password):
         """Authenticate a user against crowd. Returns brief information about
         the user."""
         url = ("%s/rest/usermanagement/%s/authentication?username=%s"
                % (self._uri, self._version, urllib.parse.quote(username)))
         body = ascii_bytes(ext_json.dumps({"value": password}))
         return self._request(url, body)
     def user_groups(self, username):
         """Retrieve a list of groups to which this user belongs."""
         url = ("%s/rest/usermanagement/%s/user/group/nested?username=%s"
                % (self._uri, self._version, urllib.parse.quote(username)))
         return self._request(url)
 class KallitheaAuthPlugin(auth_modules.KallitheaExternalAuthPlugin):
     def __init__(self):
         self._protocol_values = ["http", "https"]
     @hybrid_property
     def name(self):
         return "crowd"

0 comments (0 inline, 0 general)