text: # hgweb/common.py - Utility functions needed by hgweb_mod and hgwebdir_mod
  #
  # Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net>
  # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com>
  #
  # This software may be used and distributed according to the terms of the
  # GNU General Public License version 2 or any later version.
  
  import errno, mimetypes, os
  
  HTTP_OK = 200
  HTTP_NOT_MODIFIED = 304
  HTTP_BAD_REQUEST = 400
  HTTP_UNAUTHORIZED = 401
  HTTP_FORBIDDEN = 403
  HTTP_NOT_FOUND = 404
  HTTP_METHOD_NOT_ALLOWED = 405
  HTTP_SERVER_ERROR = 500
  
  
  def checkauthz(hgweb, req, op):
      '''Check permission for operation based on request data (including
      authentication info). Return if op allowed, else raise an ErrorResponse
      exception.'''
  
      user = req.env.get('REMOTE_USER')
  
      deny_read = hgweb.configlist('web', 'deny_read')
      if deny_read and (not user or deny_read == ['*'] or user in deny_read):
          raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized')
  
      allow_read = hgweb.configlist('web', 'allow_read')
      result = (not allow_read) or (allow_read == ['*'])
      if not (result or user in allow_read):
          raise ErrorResponse(HTTP_UNAUTHORIZED, 'read not authorized')
  
      if op == 'pull' and not hgweb.allowpull:
          raise ErrorResponse(HTTP_UNAUTHORIZED, 'pull not authorized')
      elif op == 'pull' or op is None: # op is None for interface requests
          return
  
      # enforce that you can only push using POST requests
      if req.env['REQUEST_METHOD'] != 'POST':
          msg = 'push requires POST request'
          raise ErrorResponse(HTTP_METHOD_NOT_ALLOWED, msg)
  
      # require ssl by default for pushing, auth info cannot be sniffed
      # and replayed
      scheme = req.env.get('wsgi.url_scheme')
      if hgweb.configbool('web', 'push_ssl', True) and scheme != 'https':
          raise ErrorResponse(HTTP_OK, 'ssl required')
  
      deny = hgweb.configlist('web', 'deny_push')
      if deny and (not user or deny == ['*'] or user in deny):
          raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')
  
      allow = hgweb.configlist('web', 'allow_push')
      result = allow and (allow == ['*'] or user in allow)
      if not result:
          raise ErrorResponse(HTTP_UNAUTHORIZED, 'push not authorized')
  
  # Hooks for hgweb permission checks; extensions can add hooks here.
  # Each hook is invoked like this: hook(hgweb, request, operation),
  # where operation is either read, pull or push. Hooks should either
  # raise an ErrorResponse exception, or just return.
  #
  # It is possible to do both authentication and authorization through
  # this.
  permhooks = [checkauthz]
  
  
  class ErrorResponse(Exception):
      def __init__(self, code, message=None, headers=[]):
          if message is None:
              message = _statusmessage(code)
          Exception.__init__(self)
          self.code = code
          self.message = message
          self.headers = headers
      def __str__(self):
          return self.message
  
  class continuereader(object):
      def __init__(self, f, write):
          self.f = f
          self._write = write
          self.continued = False
  
      def read(self, amt=-1):
          if not self.continued:
              self.continued = True
              self._write('HTTP/1.1 100 Continue\r\n\r\n')
          return self.f.read(amt)
  
      def __getattr__(self, attr):
          if attr in ('close', 'readline', 'readlines', '__iter__'):
              return getattr(self.f, attr)
          raise AttributeError()
  
  def _statusmessage(code):
      from BaseHTTPServer import BaseHTTPRequestHandler
      responses = BaseHTTPRequestHandler.responses
      return responses.get(code, ('Error', 'Unknown error'))[0]
  
  def statusmessage(code, message=None):
      return '%d %s' % (code, message or _statusmessage(code))
  
  def get_stat(spath):
      """stat changelog if it exists, spath otherwise"""
      cl_path = os.path.join(spath, "00changelog.i")
      if os.path.exists(cl_path):
          return os.stat(cl_path)
      else:
          return os.stat(spath)
  
  def get_mtime(spath):
      return get_stat(spath).st_mtime
  
  def staticfile(directory, fname, req):
      """return a file inside directory with guessed Content-Type header
  
      fname always uses '/' as directory separator and isn't allowed to
      contain unusual path components.
      Content-Type is guessed using the mimetypes module.
      Return an empty string if fname is illegal or file not found.
  
      """
      parts = fname.split('/')
      for part in parts:
          if (part in ('', os.curdir, os.pardir) or
              os.sep in part or os.altsep is not None and os.altsep in part):
              return ""
      fpath = os.path.join(*parts)
      if isinstance(directory, str):
          directory = [directory]
      for d in directory:
          path = os.path.join(d, fpath)
          if os.path.exists(path):
              break
      try:
          os.stat(path)
          ct = mimetypes.guess_type(path)[0] or "text/plain"
          req.respond(HTTP_OK, ct, length = os.path.getsize(path))
          fp = open(path, 'rb')
          data = fp.read()
          fp.close()
          return data
      except TypeError:
          raise ErrorResponse(HTTP_SERVER_ERROR, 'illegal filename')
      except OSError, err:
          if err.errno == errno.ENOENT:
              raise ErrorResponse(HTTP_NOT_FOUND)
          else:
              raise ErrorResponse(HTTP_SERVER_ERROR, err.strerror)
  
  def paritygen(stripecount, offset=0):
      """count parity of horizontal stripes for easier reading"""
      if stripecount and offset:
          # account for offset, e.g. due to building the list in reverse
          count = (stripecount + offset) % stripecount
          parity = (stripecount + offset) / stripecount & 1
      else:
          count = 0
          parity = 0
      while True:
          yield parity
          count += 1
          if stripecount and count >= stripecount:
              parity = 1 - parity
              count = 0
  
  def get_contact(config):
      """Return repo contact information or empty string.
  
      web.contact is the primary source, but if that is not set, try
      ui.username or $EMAIL as a fallback to display something useful.
      """
      return (config("web", "contact") or
              config("ui", "username") or
              os.environ.get("EMAIL") or "")
  
  def caching(web, req):
      tag = str(web.mtime)
      if req.env.get('HTTP_IF_NONE_MATCH') == tag:
          raise ErrorResponse(HTTP_NOT_MODIFIED)
      req.headers.append(('ETag', tag))
  