ssl support for the python client

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@708718 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 103 insertions, 39 deletions

diff --git a/qpid/python/qpid/connection.py b/qpid/python/qpid/connection.py
index ce27a74489..4c9c02822a 100644
--- a/qpid/python/qpid/connection.py
+++ b/qpid/python/qpid/connection.py
@@ -17,7 +17,7 @@
 # under the License.
 #
 
-import datatypes, session
+import datatypes, session, socket
 from threading import Thread, Condition, RLock
 from util import wait, notify
 from assembler import Assembler, Segment
@@ -44,10 +44,27 @@ def client(*args, **kwargs):
 def server(*args, **kwargs):
   return delegates.Server(*args, **kwargs)
 
+class SSLWrapper:
+
+  def __init__(self, ssl):
+    self.ssl = ssl
+
+  def recv(self, n):
+    return self.ssl.read(n)
+
+  def send(self, s):
+    return self.ssl.write(s)
+
+def sslwrap(sock):
+  if isinstance(sock, socket.SSLType):
+    return SSLWrapper(sock)
+  else:
+    return sock
+
 class Connection(Assembler):
 
   def __init__(self, sock, spec=None, delegate=client, **args):
-    Assembler.__init__(self, sock)
+    Assembler.__init__(self, sslwrap(sock))
     if spec == None:
       spec = load(default())
     self.spec = spec
diff --git a/qpid/python/qpid/qmfconsole.py b/qpid/python/qpid/qmfconsole.py
index bfd201fcc3..55f2370a28 100644
--- a/qpid/python/qpid/qmfconsole.py
+++ b/qpid/python/qpid/qmfconsole.py
@@ -27,7 +27,7 @@ import re
 from qpid.peer       import Closed
 from qpid.connection import Connection, ConnectionFailed
 from qpid.datatypes  import uuid4, Message, RangedSet
-from qpid.util       import connect
+from qpid.util       import connect, ssl, URL
 from qpid.codec010   import StringCodec as Codec
 from threading       import Lock, Condition
 from time            import time, strftime, gmtime
@@ -86,21 +86,17 @@ class Console:
     """ """
     pass
 
-class BrokerURL:
+class BrokerURL(URL):
   def __init__(self, text):
-    rex = re.compile(r"""
-    # [   <user>  [   / <password> ] @]  <host>  [   :<port>   ]
-    ^ (?: ([^/]*) (?: / ([^@]*)   )? @)? ([^:]+) (?: :([0-9]+))?$""", re.X)
-    match = rex.match(text)
-    if not match: raise ValueError("'%s' is not a valid broker url" % (text))
-    user, password, host, port = match.groups()
-
-    socket.gethostbyname(host)
-    self.host = host
-    if port: self.port = int(port)
-    else: self.port = 5672
-    self.authName = user or "guest"
-    self.authPass = password or "guest"
+    URL.__init__(self, text)
+    socket.gethostbyname(self.host)
+    if self.port is None:
+      if self.scheme == URL.AMQPS:
+        self.port = 5671
+      else:
+        self.port = 5672
+    self.authName = self.user or "guest"
+    self.authPass = self.password or "guest"
     self.authMech = "PLAIN"
 
   def name(self):
@@ -178,7 +174,8 @@ class Session:
   def addBroker(self, target="localhost"):
     """ Connect to a Qpid broker.  Returns an object of type Broker. """
     url = BrokerURL(target)
-    broker = Broker(self, url.host, url.port, url.authMech, url.authName, url.authPass)
+    broker = Broker(self, url.host, url.port, url.authMech, url.authName, url.authPass,
+                    ssl = url.scheme == URL.AMQPS)
     if not broker.isConnected and not self.manageConnections:
       raise Exception(broker.error)
 
@@ -1075,10 +1072,11 @@ class Broker:
   """ """
   SYNC_TIME = 60
 
-  def __init__(self, session, host, port, authMech, authUser, authPass):
+  def __init__(self, session, host, port, authMech, authUser, authPass, ssl=False):
     self.session  = session
     self.host     = host
     self.port     = port
+    self.ssl = ssl
     self.authUser = authUser
     self.authPass = authPass
     self.agents   = {}
@@ -1129,7 +1127,10 @@ class Broker:
   def _tryToConnect(self):
     try:
       self.amqpSessionId = "%s.%d" % (os.uname()[1], os.getpid())
-      self.conn = Connection(connect(self.host, self.port), username=self.authUser, password=self.authPass)
+      sock = connect(self.host, self.port)
+      if self.ssl:
+        sock = ssl(sock)
+      self.conn = Connection(sock, username=self.authUser, password=self.authPass)
       self.conn.start()
       self.replyName = "reply-%s" % self.amqpSessionId
       self.amqpSession = self.conn.session(self.amqpSessionId)
diff --git a/qpid/python/qpid/testlib.py b/qpid/python/qpid/testlib.py
index b8f4c29ceb..53723148b7 100644
--- a/qpid/python/qpid/testlib.py
+++ b/qpid/python/qpid/testlib.py
@@ -32,7 +32,7 @@ from qpid.message import Message
 #0-10 support
 from qpid.connection import Connection
 from qpid.spec010 import load
-from qpid.util import connect
+from qpid.util import connect, ssl, URL
 
 def findmodules(root):
     """Find potential python modules under directory root"""
@@ -70,8 +70,9 @@ Options:
   -s/--spec <spec.xml> : URL of AMQP XML specification or one of these abbreviations:
                            0-8 - use the default 0-8 specification.
                            0-9 - use the default 0-9 specification.
+                           0-10-errata - use the 0-10 specification with qpid errata.
   -e/--errata <errata.xml> : file containing amqp XML errata
-  -b/--broker [<user>[/<password>]@]<host>[:<port>] : broker to connect to
+  -b/--broker [amqps://][<user>[/<password>]@]<host>[:<port>] : broker to connect to
   -v/--verbose             : verbose - lists tests as they are run.
   -d/--debug               : enable debug logging.
   -i/--ignore <test>       : ignore the named test.
@@ -82,15 +83,20 @@ Options:
         sys.exit(1)
 
     def setBroker(self, broker):
-        rex = re.compile(r"""
-        # [   <user>  [   / <password> ] @]  <host>  [   :<port>   ]
-        ^ (?: ([^/]*) (?: / ([^@]*)   )? @)? ([^:]+) (?: :([0-9]+))?$""", re.X)
-        match = rex.match(broker)
-        if not match: self._die("'%s' is not a valid broker" % (broker))
-        self.user, self.password, self.host, self.port = match.groups()
-        self.port = int(default(self.port, 5672))
-        self.user = default(self.user, "guest")
-        self.password = default(self.password, "guest")
+        try:
+            self.url = URL(broker)
+        except ValueError:
+            self._die("'%s' is not a valid broker" % (broker))
+        self.user = default(self.url.user, "guest")
+        self.password = default(self.url.password, "guest")
+        self.host = self.url.host
+        if self.url.scheme == URL.AMQPS:
+            self.ssl = True
+            default_port = 5671
+        else:
+            self.ssl = False
+            default_port = 5672
+        self.port = default(self.url.port, default_port)
 
     def ignoreFile(self, filename):
         f = file(filename)
@@ -129,6 +135,7 @@ Options:
             if opt in ("-I", "--ignore-file"): self.ignoreFile(value)
             if opt in ("-S", "--skip-self-test"): self.skip_self_test = True
             if opt in ("-F", "--spec-folder"): TestRunner.SPEC_FOLDER = value
+
 	# Abbreviations for default settings.
         if (self.specfile == "0-10"):
             self.spec = load(self.get_spec_file("amqp.0-10.xml"))
@@ -352,20 +359,20 @@ class TestBase010(unittest.TestCase):
     """
 
     def setUp(self):
-        spec = testrunner.spec
-        self.conn = Connection(connect(testrunner.host, testrunner.port), spec,
-                               username=testrunner.user, password=testrunner.password)
-        self.conn.start(timeout=10)        
+        self.conn = self.connect()
         self.session = self.conn.session("test-session", timeout=10)
         self.qmf = None
 
     def startQmf(self):
         self.qmf = qpid.qmfconsole.Session()
-        self.qmf_broker = self.qmf.addBroker("%s:%d" % (testrunner.host, testrunner.port))
+        self.qmf_broker = self.qmf.addBroker(str(testrunner.url))
 
     def connect(self, host=None, port=None):
-        spec = testrunner.spec
-        conn = Connection(connect(host or testrunner.host, port or testrunner.port), spec)
+        sock = connect(host or testrunner.host, port or testrunner.port)
+        if testrunner.url.scheme == URL.AMQPS:
+            sock = ssl(sock)
+        conn = Connection(sock, testrunner.spec, username=testrunner.user,
+                          password=testrunner.password)
         conn.start(timeout=10)
         return conn
 
diff --git a/qpid/python/qpid/util.py b/qpid/python/qpid/util.py
index 1140cbe5ef..1ca616f1f5 100644
--- a/qpid/python/qpid/util.py
+++ b/qpid/python/qpid/util.py
@@ -17,7 +17,9 @@
 # under the License.
 #
 
-import os, socket, time, textwrap
+import os, socket, time, textwrap, re
+
+ssl = socket.ssl
 
 def connect(host, port):
   sock = socket.socket()
@@ -76,3 +78,40 @@ def fill(text, indent, heading = None):
     init = sub
   w = textwrap.TextWrapper(initial_indent = init, subsequent_indent = sub)
   return w.fill(" ".join(text.split()))
+
+class URL:
+
+  RE = re.compile(r"""
+        # [   <scheme>://  ] [    <user>   [   / <password>   ] @]   <host>   [   :<port>   ]
+        ^ (?: ([^:/@]+)://)? (?: ([^:/@]+) (?: / ([^:/@]+)   )? @)? ([^@:/]+) (?: :([0-9]+))?$
+""", re.X)
+
+  AMQPS = "amqps"
+  AMQP = "amqp"
+
+  def __init__(self, s):
+    match = URL.RE.match(s)
+    if match is None:
+      raise ValueError(s)
+    self.scheme, self.user, self.password, self.host, port = match.groups()
+    if port is None:
+      self.port = None
+    else:
+      self.port = int(port)
+
+  def __repr__(self):
+    return "URL(%r)" % str(self)
+
+  def __str__(self):
+    s = ""
+    if self.scheme:
+      s += "%s://" % self.scheme
+    if self.user:
+      s += self.user
+      if self.password:
+        s += "/%s" % self.password
+      s += "@"
+    s += self.host
+    if self.port:
+      s += ":%s" % self.port
+    return s