Menu
▾
▴
[Pywebsvcs-checkins] SF.net SVN: pywebsvcs: [1336] trunk/zsi
|
From: <bov...@us...> - 2007-01-25 23:09:16
|
Revision: 1336
http://svn.sourceforge.net/pywebsvcs/?rev=1336&view=rev
Author: boverhof
Date: 2007-01-25 15:09:17 -0800 (Thu, 25 Jan 2007)
Log Message:
-----------
A test/test_rfc2617.py
M test/test_zsi.py
M test/test_zsi_net.py
M test/wsdl2py/test_AWSECommerceService.py
M ZSI/digest_auth.py
M ZSI/client.py
-- merge over changes
Report [ 1593310 ] Http Digest authentication fails
Modification of Patch [ 1593314 ] HttpDigest authentication patch
Modified Paths:
--------------
trunk/zsi/ZSI/client.py
trunk/zsi/ZSI/digest_auth.py
trunk/zsi/test/test_zsi.py
trunk/zsi/test/test_zsi_net.py
trunk/zsi/test/wsdl2py/test_AWSECommerceService.py
Added Paths:
-----------
trunk/zsi/test/test_rfc2617.py
Modified: trunk/zsi/ZSI/client.py
===================================================================
--- trunk/zsi/ZSI/client.py 2007-01-25 22:39:50 UTC (rev 1335)
+++ trunk/zsi/ZSI/client.py 2007-01-25 23:09:17 UTC (rev 1336)
@@ -277,9 +277,9 @@
print >>self.trace, "_" * 33, time.ctime(time.time()), "REQUEST:"
print >>self.trace, soapdata
- #scheme,netloc,path,nil,nil,nil = urlparse.urlparse(url)
- path = _get_postvalue_from_absoluteURI(url)
- self.h.putrequest("POST", path)
+ url = url or self.url
+ request_uri = _get_postvalue_from_absoluteURI(url)
+ self.h.putrequest("POST", request_uri)
self.h.putheader("Content-Length", "%d" % len(soapdata))
self.h.putheader("Content-Type", 'text/xml; charset=utf-8')
self.__addcookies()
@@ -296,7 +296,7 @@
elif self.auth_style == AUTH.httpdigest and not headers.has_key('Authorization') \
and not headers.has_key('Expect'):
def digest_auth_cb(response):
- self.SendSOAPDataHTTPDigestAuth(response, soapdata, url, soapaction, **kw)
+ self.SendSOAPDataHTTPDigestAuth(response, soapdata, url, request_uri, soapaction, **kw)
self.http_callbacks[401] = None
self.http_callbacks[401] = digest_auth_cb
@@ -308,7 +308,7 @@
# Clear prior receive state.
self.data, self.ps = None, None
- def SendSOAPDataHTTPDigestAuth(self, response, soapdata, url, soapaction, **kw):
+ def SendSOAPDataHTTPDigestAuth(self, response, soapdata, url, request_uri, soapaction, **kw):
'''Resend the initial request w/http digest authorization headers.
The SOAP server has requested authorization. Fetch the challenge,
generate the authdict for building a response.
@@ -333,7 +333,7 @@
dict_fetch(chaldict,'realm',None) and \
dict_fetch(chaldict,'qop',None):
authdict = generate_response(chaldict,
- url, self.auth_user, self.auth_pass, method='POST')
+ request_uri, self.auth_user, self.auth_pass, method='POST')
headers = {\
'Authorization':build_authorization_arg(authdict),
'Expect':'100-continue',
Modified: trunk/zsi/ZSI/digest_auth.py
===================================================================
--- trunk/zsi/ZSI/digest_auth.py 2007-01-25 22:39:50 UTC (rev 1335)
+++ trunk/zsi/ZSI/digest_auth.py 2007-01-25 23:09:17 UTC (rev 1336)
@@ -2,7 +2,7 @@
# $Header$
'''Utilities for HTTP Digest Authentication
'''
-
+import re
from md5 import md5
import random
import time
@@ -75,24 +75,31 @@
def fetch_challenge(http_header):
+ """ apparently keywords Basic and Digest are not being checked
+ anywhere and decisions are being made based on authorization
+ configuration of client, so I guess you better know what you are
+ doing. Here I am requiring one or the other be specified.
+
+ challenge Basic auth_param
+ challenge Digest auth_param
"""
- Create a challenge dictionary from a HTTPResponse objects getheaders() method.
- """
- chaldict = {}
- vals = http_header.split(' ')
- chaldict['challenge'] = vals[0]
- for val in vals[1:]:
- try:
- a,b = val.split('=')
- b=b.replace('"','')
- b=b.replace("'",'')
- b=b.replace(",",'')
- chaldict[a.lower()] = b
- except:
- pass
- return chaldict
+ m = fetch_challenge.wwwauth_header_re.match(http_header)
+ if m is None:
+ raise RuntimeError, 'expecting "WWW-Authenticate header [Basic,Digest]"'
+ d = dict(challenge=m.groups()[0])
+ m = fetch_challenge.auth_param_re.search(http_header)
+ while m is not None:
+ k,v = http_header[m.start():m.end()].split('=')
+ d[k.lower()] = v[1:-1]
+ m = fetch_challenge.auth_param_re.search(http_header, m.end())
+
+ return d
+fetch_challenge.wwwauth_header_re = re.compile(r'\s*([bB]asic|[dD]igest)\s+(?:[\w]+="[^"]+",?\s*)?')
+fetch_challenge.auth_param_re = re.compile(r'[\w]+="[^"]+"')
+
+
def build_authorization_arg(authdict):
"""
Create an "Authorization" header value from an authdict (created by generate_response()).
Added: trunk/zsi/test/test_rfc2617.py
===================================================================
--- trunk/zsi/test/test_rfc2617.py (rev 0)
+++ trunk/zsi/test/test_rfc2617.py 2007-01-25 23:09:17 UTC (rev 1336)
@@ -0,0 +1,52 @@
+#!/usr/bin/env python
+"""
+RFC2617
+
+HTTP Authentication: Basic and Digest Access Authentication
+"""
+import unittest
+from ZSI import digest_auth
+from ZSI.wstools.logging import setBasicLoggerDEBUG
+setBasicLoggerDEBUG()
+
+class DATestCase(unittest.TestCase):
+ "test Union TypeCode"
+
+ def check_challenge_single_www_authenticate_header(self):
+ challenge='Basic realm="WallyWorld"'
+ print "=="*30
+ print challenge
+ print "=="*30
+ cd = digest_auth.fetch_challenge(challenge)
+ expect = {'challenge': 'Basic', 'realm': 'WallyWorld'}
+ self.failUnless(cd == expect, 'Expected equivalent')
+
+ def check_challenge_single_www_authenticate_header2(self):
+ challenge='Basic realm="Wally World"'
+ cd = digest_auth.fetch_challenge(challenge)
+ expect = {'challenge': 'Basic', 'realm': 'Wally World'}
+ self.failUnless(cd == expect, 'Expected equivalent')
+
+ def check_challenge_single_www_authenticate_header3(self):
+ challenge = '''Digest
+ realm="tes...@ho...",
+ qop="auth,auth-int",
+ nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093",
+ opaque="5ccc069c403ebaf9f0171e9517f40e41"'''
+ cd = digest_auth.fetch_challenge(challenge)
+ expect = {'nonce': 'dcd98b7102dd2f0e8b11d0f600bfb0c093', 'challenge': 'Digest', 'opaque': '5ccc069c403ebaf9f0171e9517f40e41', 'realm': 'tes...@ho...', 'qop': 'auth,auth-int'}
+ self.failUnless(cd == expect, 'Expected equivalent')
+
+
+
+def makeTestSuite():
+ suite = unittest.TestSuite()
+ suite.addTest(unittest.makeSuite(DATestCase, "check"))
+ return suite
+
+def main():
+ unittest.main(defaultTest="makeTestSuite")
+
+if __name__ == '__main__':
+ main()
+
Modified: trunk/zsi/test/test_zsi.py
===================================================================
--- trunk/zsi/test/test_zsi.py 2007-01-25 22:39:50 UTC (rev 1335)
+++ trunk/zsi/test/test_zsi.py 2007-01-25 23:09:17 UTC (rev 1336)
@@ -11,6 +11,7 @@
import test_union
import test_list
import test_TCtimes
+import test_rfc2617
def makeTestSuite():
return unittest.TestSuite(
Modified: trunk/zsi/test/test_zsi_net.py
===================================================================
--- trunk/zsi/test/test_zsi_net.py 2007-01-25 22:39:50 UTC (rev 1335)
+++ trunk/zsi/test/test_zsi_net.py 2007-01-25 23:09:17 UTC (rev 1336)
@@ -12,6 +12,7 @@
import test_union
import test_TCtimes
import test_list
+import test_rfc2617
def makeTestSuite():
return unittest.TestSuite(
Modified: trunk/zsi/test/wsdl2py/test_AWSECommerceService.py
===================================================================
--- trunk/zsi/test/wsdl2py/test_AWSECommerceService.py 2007-01-25 22:39:50 UTC (rev 1335)
+++ trunk/zsi/test/wsdl2py/test_AWSECommerceService.py 2007-01-25 23:09:17 UTC (rev 1336)
@@ -47,9 +47,9 @@
"""Test case for Amazon ECommerce Web service
"""
name = "test_AWSECommerceService"
- client_file_name = "AWSECommerceService_services.py"
- types_file_name = "AWSECommerceService_services_types.py"
- server_file_name = "AWSECommerceService_services_server.py"
+ client_file_name = "AWSECommerceService_client.py"
+ types_file_name = "AWSECommerceService_types.py"
+ server_file_name = "AWSECommerceService_server.py"
def __init__(self, methodName):
ServiceTestCase.__init__(self, methodName)
@@ -96,7 +96,7 @@
def test_net_ItemSearch(self):
loc = self.client_module.AWSECommerceServiceLocator()
- port = loc.getAWSECommerceServicePortType(**self.getPortKWArgs())
+ port = loc.getAWSECommerceServicePort(**self.getPortKWArgs())
msg = self.client_module.ItemSearchRequestMsg()
msg.SubscriptionId = '0HP1WHME000749APYWR2'
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
