sys/lib/python/hgext/hgfactotum.py

   1 ''' factotum support '''
   2
   3 import mercurial.url
   4 import urllib2
   5 import factotum
   6 import base64
   7
   8 class factotumbasic(urllib2.BaseHandler):
   9         def __init__(self, passmgr=None):
  10                 self.f = factotum.Factotum()
  11                 self.retried = 0
  12                 self.auth = None
  13         def http_error_401(self, req, fp, code, msg, headers):
  14                 host = urllib2.urlparse.urlparse(req.get_full_url())[1]
  15                 authreq = headers.get('www-authenticate', None)
  16                 if authreq == None: return None
  17                 authreq = authreq.split(' ', 1)
  18                 if authreq[0].lower() != 'basic': return None
  19                 chal = urllib2.parse_keqv_list(urllib2.parse_http_list(authreq[1]))
  20                 realm = chal['realm']
  21                 self.auth = (host, realm)
  22                 self.retried += 1
  23                 if self.retried >= 3:
  24                         self.f.delkey(proto="pass", host=host, realm=realm, role="client")
  25                 self.f.start(proto="pass", host=host, realm=realm, role="client")
  26                 pw = self.f.read().replace(' ', ':', 1)
  27                 val = 'Basic %s' % base64.b64encode(pw).strip()
  28                 if req.headers.get('Authorization', None) == val: return None
  29                 req.add_header('Authorization', val)
  30                 result = self.parent.open(req)
  31                 self.retried = 0
  32                 return result
  33         def http_error_403(self, req, fp, code, msg, headers):
  34                 if self.auth != None:
  35                         self.f.delkey(proto="pass", host=self.auth[0], realm=self.auth[1], role="client")
  36                         self.auth = None
  37
  38 class factotumdigest(urllib2.BaseHandler):
  39         auth_header = 'Authorization'
  40         handler_order = 490
  41
  42         def __init__(self, passmgr=None):
  43                 self.f = factotum.Factotum()
  44                 self.retried = 0
  45         def http_error_401(self, req, fp, code, msg, headers):
  46                 self.retried += 1
  47                 host = urllib2.urlparse.urlparse(req.get_full_url())[1]
  48                 authreq = headers.get('www-authenticate', None)
  49                 if authreq == None: return None
  50                 authreq = authreq.split(' ', 1)
  51                 if authreq[0].lower() != 'digest': return None
  52                 chal = urllib2.parse_keqv_list(urllib2.parse_http_list(authreq[1]))
  53                 realm = chal['realm']
  54                 nonce = chal['nonce']
  55                 if self.retried >= 6:
  56                         self.f.delkey(proto="httpdigest", realm=realm, host=host)
  57                 self.f.start(proto="httpdigest", role="client", realm=realm, host=host)
  58                 self.f.write(nonce + ' ' + req.get_method() + ' ' + req.get_selector())
  59                 resp = self.f.read()
  60                 user = self.f.attr()["user"]
  61                 self.f.close()
  62                 val = 'Digest username="%s", realm="%s", nonce="%s", uri="%s", response="%s", algorithm=MD5' % (user, realm, nonce, req.get_selector(), resp)
  63                 if req.headers.get('Authorization', None) == val: return None
  64                 req.add_unredirected_header('Authorization', val)
  65                 result = self.parent.open(req)
  66                 self.retried = 0
  67                 return result
  68
  69 urllib2.HTTPBasicAuthHandler = factotumbasic
  70 mercurial.url.httpdigestauthhandler = factotumdigest