kallithea Changeset - 26bc2f02d9cd

Changeset - 26bc2f02d9cd

Parent rev.

Child rev.

[Not reviewed]

default

0 3 0

Mads Kiilerich - 9 years ago 2017-01-22 01:16:52
mads@kiilerich.com

lib: the variable 'request' is usually used for the pylons request object - avoid using it for other purposes

3 files changed with 14 insertions and 14 deletions:

kallithea/lib/auth_modules/auth_crowd.py

kallithea/lib/middleware/pygrack.py

kallithea/lib/recaptcha.py

0 comments (0 inline, 0 general)

kallithea/lib/auth_modules/auth_crowd.py

➞

Show inline comments

@@ @@ -43,104 +43,104 @@ class CrowdServer(object): @@
         on the given port, and using the given method (https/http). user and
         passwd can be set here or with set_credentials. If unspecified,
         "version" defaults to "latest".
         example::
             cserver = CrowdServer(host="127.0.0.1",
                                   port="8095",
                                   user="some_app",
                                   passwd="some_passwd",
                                   version="1")
         """
         if not "port" 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 = urllib2.HTTPPasswordMgrWithDefaultRealm()
         mgr.add_password(None, self._uri, self.user, self.passwd)
         handler = urllib2.HTTPBasicAuthHandler(mgr)
         self.opener = urllib2.build_opener(handler)
     def _request(self, url, body=None, headers=None,
                  method=None, noformat=False,
                  empty_response_ok=False):
         _headers = {"Content-type": "application/json",
                     "Accept": "application/json"}
         if self.user and self.passwd:
             authstring = base64.b64encode("%s:%s" % (self.user, self.passwd))
             _headers["Authorization"] = "Basic %s" % authstring
         if headers:
             _headers.update(headers)
         log.debug("Sent crowd: \n%s",
                   formatted_json({"url": url, "body": body,
                                            "headers": _headers}))
-        request = urllib2.Request(url, body, _headers)
+        req = urllib2.Request(url, body, _headers)
         if method:
-            request.get_method = lambda: method
+            req.get_method = lambda: method
         global msg
         msg = ""
         try:
-            rdoc = self.opener.open(request)
+            rdoc = self.opener.open(req)
             msg = "".join(rdoc.readlines())
             if not msg and empty_response_ok:
                 rval = {}
                 rval["status"] = True
                 rval["error"] = "Response body was empty"
             elif not noformat:
                 rval = json.loads(msg)
                 rval["status"] = True
             else:
                 rval = "".join(rdoc.readlines())
         except Exception as e:
             if not noformat:
                 rval = {"status": False,
                         "body": body,
                         "error": str(e) + "\n" + 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, urllib2.quote(username)))
         body = 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, urllib2.quote(username)))
         return self._request(url)
 class KallitheaAuthPlugin(auth_modules.KallitheaExternalAuthPlugin):
     @hybrid_property
     def name(self):
         return "crowd"
     def settings(self):
         settings = [
+            {
                 "name": "host",
                 "validator": self.validators.UnicodeString(strip=True),
                 "type": "string",
                 "description": "The FQDN or IP of the Atlassian CROWD Server",
                 "default": "127.0.0.1",

kallithea/lib/middleware/pygrack.py

➞

Show inline comments

@@ @@ -46,185 +46,185 @@ class FileWrapper(object): @@
         self.content_length = content_length
         self.remain = content_length
     def read(self, size):
         if size <= self.remain:
             try:
                 data = self.fd.read(size)
             except socket.error:
                 raise IOError(self)
             self.remain -= size
         elif self.remain:
             data = self.fd.read(self.remain)
             self.remain = 0
         else:
             data = None
         return data
     def __repr__(self):
         return '<FileWrapper %s len: %s, read: %s>' % (
             self.fd, self.content_length, self.content_length - self.remain
+        )
 class GitRepository(object):
     git_folder_signature = set(['config', 'head', 'info', 'objects', 'refs'])
     commands = ['git-upload-pack', 'git-receive-pack']
     def __init__(self, repo_name, content_path, extras):
         files = set([f.lower() for f in os.listdir(content_path)])
         if  not (self.git_folder_signature.intersection(files)
                 == self.git_folder_signature):
             raise OSError('%s missing git signature' % content_path)
         self.content_path = content_path
         self.valid_accepts = ['application/x-%s-result' %
                               c for c in self.commands]
         self.repo_name = repo_name
         self.extras = extras
     def _get_fixedpath(self, path):
         """
         Small fix for repo_path
         :param path:
         """
         path = safe_unicode(path)
         assert path.startswith('/' + self.repo_name + '/')
         return path[len(self.repo_name) + 2:].strip('/')
-    def inforefs(self, request, environ):
+    def inforefs(self, req, environ):
         """
         WSGI Response producer for HTTP GET Git Smart
         HTTP /info/refs request.
         """
-        git_command = request.GET.get('service')
+        git_command = req.GET.get('service')
         if git_command not in self.commands:
             log.debug('command %s not allowed', git_command)
             return exc.HTTPMethodNotAllowed()
         # From Documentation/technical/http-protocol.txt shipped with Git:
+        #
         # Clients MUST verify the first pkt-line is `# service=$servicename`.
         # Servers MUST set $servicename to be the request parameter value.
         # Servers SHOULD include an LF at the end of this line.
         # Clients MUST ignore an LF at the end of the line.
+        #
         #  smart_reply     =  PKT-LINE("# service=$servicename" LF)
         #                     ref_list
         #                     "0000"
         server_advert = '# service=%s\n' % git_command
         packet_len = str(hex(len(server_advert) + 4)[2:].rjust(4, '0')).lower()
         _git_path = kallithea.CONFIG.get('git_path', 'git')
         cmd = [_git_path, git_command[4:],
                '--stateless-rpc', '--advertise-refs', self.content_path]
         log.debug('handling cmd %s', cmd)
         try:
             out = subprocessio.SubprocessIOChunker(cmd,
                 starting_values=[packet_len + server_advert + '0000']
+            )
         except EnvironmentError as e:
             log.error(traceback.format_exc())
             raise exc.HTTPExpectationFailed()
         resp = Response()
         resp.content_type = 'application/x-%s-advertisement' % str(git_command)
         resp.charset = None
         resp.app_iter = out
         return resp
-    def backend(self, request, environ):
+    def backend(self, req, environ):
         """
         WSGI Response producer for HTTP POST Git Smart HTTP requests.
         Reads commands and data from HTTP POST's body.
         returns an iterator obj with contents of git command's
         response to stdout
         """
         _git_path = kallithea.CONFIG.get('git_path', 'git')
-        git_command = self._get_fixedpath(request.path_info)
+        git_command = self._get_fixedpath(req.path_info)
         if git_command not in self.commands:
             log.debug('command %s not allowed', git_command)
             return exc.HTTPMethodNotAllowed()
         if 'CONTENT_LENGTH' in environ:
             inputstream = FileWrapper(environ['wsgi.input'],
-                                      request.content_length)
+                                      req.content_length)
         else:
             inputstream = environ['wsgi.input']
         gitenv = dict(os.environ)
         # forget all configs
         gitenv['GIT_CONFIG_NOGLOBAL'] = '1'
         cmd = [_git_path, git_command[4:], '--stateless-rpc', self.content_path]
         log.debug('handling cmd %s', cmd)
         try:
             out = subprocessio.SubprocessIOChunker(
                 cmd,
                 inputstream=inputstream,
                 env=gitenv,
                 cwd=self.content_path,
+            )
         except EnvironmentError as e:
             log.error(traceback.format_exc())
             raise exc.HTTPExpectationFailed()
         if git_command in [u'git-receive-pack']:
             # updating refs manually after each push.
             # Needed for pre-1.7.0.4 git clients using regular HTTP mode.
             from kallithea.lib.vcs import get_repo
             from dulwich.server import update_server_info
             repo = get_repo(self.content_path)
             if repo:
                 update_server_info(repo._repo)
         resp = Response()
         resp.content_type = 'application/x-%s-result' % git_command.encode('utf8')
         resp.charset = None
         resp.app_iter = out
         return resp
     def __call__(self, environ, start_response):
         request = Request(environ)
         _path = self._get_fixedpath(request.path_info)
         req = Request(environ)
         _path = self._get_fixedpath(req.path_info)
         if _path.startswith('info/refs'):
             app = self.inforefs
-        elif [a for a in self.valid_accepts if a in request.accept]:
+        elif [a for a in self.valid_accepts if a in req.accept]:
             app = self.backend
         try:
-            resp = app(request, environ)
+            resp = app(req, environ)
         except exc.HTTPException as e:
             resp = e
             log.error(traceback.format_exc())
         except Exception as e:
             log.error(traceback.format_exc())
             resp = exc.HTTPInternalServerError()
         return resp(environ, start_response)
 class GitDirectory(object):
     def __init__(self, repo_root, repo_name, extras):
         repo_location = os.path.join(repo_root, repo_name)
         if not os.path.isdir(repo_location):
             raise OSError(repo_location)
         self.content_path = repo_location
         self.repo_name = repo_name
         self.repo_location = repo_location
         self.extras = extras
     def __call__(self, environ, start_response):
         content_path = self.content_path
         try:
             app = GitRepository(self.repo_name, content_path, self.extras)
         except (AssertionError, OSError):
             content_path = os.path.join(content_path, '.git')
             if os.path.isdir(content_path):
                 app = GitRepository(self.repo_name, content_path, self.extras)
             else:
                 return exc.HTTPNotFound()(environ, start_response)
         return app(environ, start_response)
 def make_wsgi_app(repo_name, repo_root, extras):
     from dulwich.web import LimitedInputFilter, GunzipFilter
     app = GitDirectory(repo_root, repo_name, extras)
     return GunzipFilter(LimitedInputFilter(app))

kallithea/lib/recaptcha.py

➞

Show inline comments

@@ @@ -31,68 +31,68 @@ def displayhtml(public_key, use_ssl=Fals @@
         server = API_SSL_SERVER
     else:
         server = API_SERVER
     return """<script type="text/javascript" src="%(ApiServer)s/challenge?k=%(PublicKey)s%(ErrorParam)s"></script>
 <noscript>
   <iframe src="%(ApiServer)s/noscript?k=%(PublicKey)s%(ErrorParam)s" height="300" width="500" frameborder="0"></iframe><br />
   <textarea name="recaptcha_challenge_field" rows="3" cols="40"></textarea>
   <input type='hidden' name='recaptcha_response_field' value='manual_challenge' />
 </noscript>
 """ % {
         'ApiServer': server,
         'PublicKey': public_key,
         'ErrorParam': error_param,
+    }
 def submit(recaptcha_challenge_field, recaptcha_response_field, private_key,
            remoteip):
     """
     Submits a reCAPTCHA request for verification. Returns RecaptchaResponse
     for the request
     recaptcha_challenge_field -- The value of recaptcha_challenge_field from the form
     recaptcha_response_field -- The value of recaptcha_response_field from the form
     private_key -- your reCAPTCHA private key
     remoteip -- the user's IP address
     """
     if not (recaptcha_response_field and recaptcha_challenge_field and
                 len(recaptcha_response_field) and len(
             recaptcha_challenge_field)):
         return RecaptchaResponse(is_valid=False,
                                  error_code='incorrect-captcha-sol')
     def encode_if_necessary(s):
         if isinstance(s, unicode):
             return s.encode('utf-8')
         return s
     params = urllib.urlencode({
         'privatekey': encode_if_necessary(private_key),
         'remoteip': encode_if_necessary(remoteip),
         'challenge': encode_if_necessary(recaptcha_challenge_field),
         'response': encode_if_necessary(recaptcha_response_field),
     })
-    request = urllib2.Request(
+    req = urllib2.Request(
         url="http://%s/recaptcha/api/verify" % VERIFY_SERVER,
         data=params,
         headers={
             "Content-type": "application/x-www-form-urlencoded",
             "User-agent": "reCAPTCHA Python"
+        }
+    )
-    httpresp = urllib2.urlopen(request)
+    httpresp = urllib2.urlopen(req)
     return_values = httpresp.read().splitlines()
     httpresp.close()
     return_code = return_values[0]
     if return_code == "true":
         return RecaptchaResponse(is_valid=True)
     else:
         return RecaptchaResponse(is_valid=False, error_code=return_values[1])

0 comments (0 inline, 0 general)