Ticket #3461: issue.3461.patch

twisted/web/error.py

@@ -185 +185 @@
 
 
 
+class SecurityIssue(Exception):
+    """
+    The requested action would result in secure information leaking.
+    """
+
+
 from twisted.web import resource as _resource
 
 class ErrorPage(_resource.ErrorPage):

twisted/web/server.py

@@ -34 +34 @@
 from twisted.python import log, reflect, failure, components
 from twisted import copyright
 from twisted.web import util as webutil, resource
-from twisted.web.error import UnsupportedMethod
+from twisted.web.error import UnsupportedMethod, SecurityIssue
 
 # backwards compatability
 date_time_string = http.datetimeToString
@@ -53 +53 @@
         return tuple(addr)
 
 class Request(pb.Copyable, http.Request, components.Componentized):
+    """
+    An HTTP request.
+
+    @ivar session: This stores a session available to HTTP and HTTPS requests.
+    @ivar secure_session: This stores a session only available to HTTPS
+            requests.
+    """
     implements(iweb.IRequest)
 
     site = None
@@ -264 +271 @@
     ### these calls remain local
 
     session = None
+    secure_session = None
+
+    def getSession(self, sessionInterface=None, secure=False):
+        """
+        Check if there is a session cookie, and if not, create it.
+
+        By default this session is available on HTTP and HTTPS requests. Set
+        L{secure} = True if you want a session that's only available on HTTPS.
+
+        If you try to create a secure session on a non-secure page, this will
+        raise a L{twisted.web.error.SecurityIssue}.
+        """
+        # Make sure we aren't creating a secure session on a non-secure page
+        if secure and not self.isSecure():
+            raise SecurityIssue('Cannot create secure session on insecure page')
+
+        cookie_string = ''
+        session = None
+
+        if not secure:
+            cookie_string = 'TWISTED_SESSION'
+            session = self.session
+
+        else:
+            cookie_string = 'TWISTED_SECURE_SESSION'
+            session = self.secure_session
 
-    def getSession(self, sessionInterface = None):
         # Session management
-        if not self.session:
-            cookiename = string.join(['TWISTED_SESSION'] + self.sitepath, "_")
+        if not session:
+            cookiename = string.join([cookie_string] + self.sitepath, "_")
             sessionCookie = self.getCookie(cookiename)
             if sessionCookie:
                 try:
-                    self.session = self.site.getSession(sessionCookie)
+                    session = self.site.getSession(sessionCookie)
                 except KeyError:
                     pass
             # if it still hasn't been set, fix it up.
-            if not self.session:
-                self.session = self.site.makeSession()
-                self.addCookie(cookiename, self.session.uid, path='/')
-        self.session.touch()
+            if not session:
+                session = self.site.makeSession()
+                self.addCookie(cookiename, session.uid, path='/',
+                               secure=secure)
+
+        session.touch()
+
+        # Save the session to the proper place
+        if not secure:
+            self.session = session
+        else:
+            self.secure_session = session
+
         if sessionInterface:
-            return self.session.getComponent(sessionInterface)
-        return self.session
+            return session.getComponent(sessionInterface)
+
+        return session
 
     def _prePathURL(self, prepath):
         port = self.getHost().port

twisted/web/test/test_web.py

@@ -17 +17 @@
 from twisted.web import server, resource, util
 from twisted.internet import defer, interfaces, task
 from twisted.web import iweb, http, http_headers
+from twisted.web.error import SecurityIssue
 from twisted.python import log
 
 
@@ -85 +86 @@
         """
         return self.headers.get(name.lower(), None)
 
-
     def setHeader(self, name, value):
         """TODO: make this assert on write() if the header is content-length
         """
         self.outgoingHeaders[name.lower()] = value
 
-    def getSession(self):
-        if self.session:
-            return self.session
-        assert not self.written, "Session cannot be requested after data has been written."
-        self.session = self.protoSession
-        return self.session
-
-
     def render(self, resource):
         """
         Render the given resource as a response to this request.
@@ -568 +560 @@
         self.assertTrue(
             verifyObject(iweb.IRequest, server.Request(DummyChannel(), True)))
 
-
-    def testChildLink(self):
+    def test_childLink(self):
         request = server.Request(DummyChannel(), 1)
         request.gotLength(0)
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
@@ -579 +570 @@
         request.requestReceived('GET', '/foo/bar/', 'HTTP/1.0')
         self.assertEqual(request.childLink('baz'), 'baz')
 
-    def testPrePathURLSimple(self):
+    def test_prePathURLSimple(self):
         request = server.Request(DummyChannel(), 1)
         request.gotLength(0)
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
         request.setHost('example.com', 80)
         self.assertEqual(request.prePathURL(), 'http://example.com/foo/bar')
 
-    def testPrePathURLNonDefault(self):
+    def test_prePathURLNonDefault(self):
         d = DummyChannel()
         d.transport.port = 81
         request = server.Request(d, 1)
@@ -595 +586 @@
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
         self.assertEqual(request.prePathURL(), 'http://example.com:81/foo/bar')
 
-    def testPrePathURLSSLPort(self):
+    def test_prePathURLSSLPort(self):
         d = DummyChannel()
         d.transport.port = 443
         request = server.Request(d, 1)
@@ -604 +595 @@
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
         self.assertEqual(request.prePathURL(), 'http://example.com:443/foo/bar')
 
-    def testPrePathURLSSLPortAndSSL(self):
+    def test_prePathURLSSLPortAndSSL(self):
         d = DummyChannel()
         d.transport = DummyChannel.SSL()
         d.transport.port = 443
@@ -614 +605 @@
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
         self.assertEqual(request.prePathURL(), 'https://example.com/foo/bar')
 
-    def testPrePathURLHTTPPortAndSSL(self):
+    def test_prePathURLHTTPPortAndSSL(self):
         d = DummyChannel()
         d.transport = DummyChannel.SSL()
         d.transport.port = 80
@@ -624 +615 @@
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
         self.assertEqual(request.prePathURL(), 'https://example.com:80/foo/bar')
 
-    def testPrePathURLSSLNonDefault(self):
+    def test_prePathURLSSLNonDefault(self):
         d = DummyChannel()
         d.transport = DummyChannel.SSL()
         d.transport.port = 81
@@ -634 +625 @@
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
         self.assertEqual(request.prePathURL(), 'https://example.com:81/foo/bar')
 
-    def testPrePathURLSetSSLHost(self):
+    def test_prePathURLSetSSLHost(self):
         d = DummyChannel()
         d.transport.port = 81
         request = server.Request(d, 1)
@@ -643 +634 @@
         request.requestReceived('GET', '/foo/bar', 'HTTP/1.0')
         self.assertEqual(request.prePathURL(), 'https://foo.com:81/foo/bar')
 
-
     def test_prePathURLQuoting(self):
         """
         L{Request.prePathURL} quotes special characters in the URL segments to
@@ -656 +646 @@
         request.requestReceived('GET', '/foo%2Fbar', 'HTTP/1.0')
         self.assertEqual(request.prePathURL(), 'http://example.com/foo%2Fbar')
 
+    def test_sessionDifferentFromSecureSession(self):
+        """
+        Ensure L{Request.session} and L{Request.secure_session} are different.
+        """
+        d = DummyChannel()
+        d.transport = DummyChannel.SSL()
+        request = server.Request(d, 1)
+        request.site = server.Site('/')
+        request.sitepath = []
+        session = request.getSession()
+        secure_session = request.getSession(secure=True)
+
+        # Check that the sessions are not None
+        self.assertTrue(session != None)
+        self.assertTrue(secure_session != None)
+
+        # Check that the sessions are different
+        self.assertNotEqual(session.uid, secure_session.uid)
+
+        # Check that the sessions are getting saved
+        self.assertEqual(session, request.session)
+        self.assertEqual(secure_session, request.secure_session)
+
+        session.expire()
+        secure_session.expire()
 
+    def test_secureSessionRaiseExceptionOnInsecureRequest(self):
+        """
+        Ensure L{Request.getSession} raises L{error.SecurityIssue} if
+        secure = True but the request is on an insecure page.
+        """
+        d = DummyChannel()
+        request = server.Request(d, 1)
+        request.site = server.Site('/')
+        request.sitepath = []
+        self.assertRaises(SecurityIssue, request.getSession, secure=True)
 
 class RootResource(resource.Resource):
     isLeaf=0