Ticket #13751: openredirect.diff

django/http/__init__.py

@@ -1 +1 @@
 import os
 import re
+import logging
 from Cookie import BaseCookie, SimpleCookie, CookieError
 from pprint import pformat
 from urllib import urlencode
-from urlparse import urljoin
+from urlparse import urljoin, urlparse
 try:
     # The mod_python version is more efficient, so try importing it first.
     from mod_python.util import parse_qsl
@@ -189 +190 @@
         for key, value in dict.items(self):
             dict.__setitem__(result, copy.deepcopy(key, memo), copy.deepcopy(value, memo))
         return result
-    
+
     def setlist(self, key, list_):
         self._assert_mutable()
         key = str_to_unicode(key, self.encoding)
@@ -432 +433 @@
 class HttpResponseRedirect(HttpResponse):
     status_code = 302
 
-    def __init__(self, redirect_to):
+    def __init__(self, redirect_to, whitelist=[], fallback_to=None):
         HttpResponse.__init__(self)
         self['Location'] = iri_to_uri(redirect_to)
 
+        if urlparse(self['Location']).scheme:
+            effective_whitelist = whitelist + getattr(settings, 'REDIRECT_WHITELIST', [])
+
+            matched = False
+
+            for pattern in effective_whitelist:
+                matched = re.compile(pattern).match(self['Location'])
+
+                if matched:
+                    break
+
+            if not matched:
+                logging.warn("found open redirect attack to %s", self['Location'])
+
+                self['Location'] = fallback_to or settings.LOGIN_REDIRECT_URL
+
 class HttpResponsePermanentRedirect(HttpResponse):
     status_code = 301
 
@@ -493 +510 @@
         return unicode(s, encoding, 'replace')
     else:
         return s
-

tests/regressiontests/httpwrappers/tests.py

@@ -1 +1 @@
 import copy
 import pickle
 import unittest
-from django.http import QueryDict, HttpResponse, CompatCookie, BadHeaderError
+from django.conf import settings
+from django.http import QueryDict, HttpResponse, CompatCookie, BadHeaderError, HttpResponseRedirect
 
 class QueryDictTests(unittest.TestCase):
     def test_missing_key(self):
@@ -17 +18 @@
         self.assertRaises(AttributeError, q.pop, 'foo')
         self.assertRaises(AttributeError, q.popitem)
         self.assertRaises(AttributeError, q.clear)
-        
+
     def test_immutable_get_with_default(self):
         q = QueryDict('')
         self.assertEqual(q.get('foo', 'default'), 'default')
@@ -34 +35 @@
         self.assertEqual(q.values(), [])
         self.assertEqual(len(q), 0)
         self.assertEqual(q.urlencode(), '')
-        
+
     def test_single_key_value(self):
         """Test QueryDict with one key/value pair"""
 
@@ -47 +48 @@
         self.assertEqual(q.get('bar', 'default'), 'default')
         self.assertEqual(q.getlist('foo'), ['bar'])
         self.assertEqual(q.getlist('bar'), [])
-        
+
         self.assertRaises(AttributeError, q.setlist, 'foo', ['bar'])
         self.assertRaises(AttributeError, q.appendlist, 'foo', ['bar'])
 
@@ -67 +68 @@
         self.assertRaises(AttributeError, q.popitem)
         self.assertRaises(AttributeError, q.clear)
         self.assertRaises(AttributeError, q.setdefault, 'foo', 'bar')
-        
+
         self.assertEqual(q.urlencode(), 'foo=bar')
-        
+
     def test_mutable_copy(self):
         """A copy of a QueryDict is mutable."""
         q = QueryDict('').copy()
         self.assertRaises(KeyError, q.__getitem__, "foo")
         q['name'] = 'john'
         self.assertEqual(q['name'], 'john')
-        
+
     def test_mutable_delete(self):
         q = QueryDict('').copy()
         q['name'] = 'john'
@@ -126 +127 @@
         """Test QueryDict with two key/value pairs with same keys."""
 
         q = QueryDict('vote=yes&vote=no')
-        
+
         self.assertEqual(q['vote'], u'no')
         self.assertRaises(AttributeError, q.__setitem__, 'something', 'bar')
-                
+
         self.assertEqual(q.get('vote', 'default'), u'no')
         self.assertEqual(q.get('foo', 'default'), 'default')
         self.assertEqual(q.getlist('vote'), [u'yes', u'no'])
         self.assertEqual(q.getlist('foo'), [])
-        
+
         self.assertRaises(AttributeError, q.setlist, 'foo', ['bar', 'baz'])
         self.assertRaises(AttributeError, q.setlist, 'foo', ['bar', 'baz'])
         self.assertRaises(AttributeError, q.appendlist, 'foo', ['bar'])
 
-        self.assertEqual(q.has_key('vote'), True)        
+        self.assertEqual(q.has_key('vote'), True)
         self.assertEqual('vote' in q, True)
         self.assertEqual(q.has_key('foo'), False)
         self.assertEqual('foo' in q, False)
@@ -148 +149 @@
         self.assertEqual(q.keys(), [u'vote'])
         self.assertEqual(q.values(), [u'no'])
         self.assertEqual(len(q), 1)
-        
+
         self.assertRaises(AttributeError, q.update, {'foo': 'bar'})
         self.assertRaises(AttributeError, q.pop, 'foo')
         self.assertRaises(AttributeError, q.popitem)
         self.assertRaises(AttributeError, q.clear)
         self.assertRaises(AttributeError, q.setdefault, 'foo', 'bar')
         self.assertRaises(AttributeError, q.__delitem__, 'vote')
-        
+
     def test_invalid_input_encoding(self):
         """
         QueryDicts must be able to handle invalid input encoding (in this
         case, bad UTF-8 encoding).
         """
         q = QueryDict('foo=bar&foo=\xff')
-        self.assertEqual(q['foo'], u'\ufffd')        
+        self.assertEqual(q['foo'], u'\ufffd')
         self.assertEqual(q.getlist('foo'), [u'bar', u'\ufffd'])
-   
+
     def test_pickle(self):
         q = QueryDict('')
         q1 = pickle.loads(pickle.dumps(q, 2))
@@ -172 +173 @@
         q = QueryDict('a=b&c=d')
         q1 = pickle.loads(pickle.dumps(q, 2))
         self.assertEqual(q == q1, True)
-        q = QueryDict('a=b&c=d&a=1') 
+        q = QueryDict('a=b&c=d&a=1')
         q1 = pickle.loads(pickle.dumps(q, 2))
         self.assertEqual(q == q1 , True)
 
@@ -181 +182 @@
         x = QueryDict("a=1&a=2", mutable=True)
         y = QueryDict("a=3&a=4")
         x.update(y)
-        self.assertEqual(x.getlist('a'), [u'1', u'2', u'3', u'4'])    
+        self.assertEqual(x.getlist('a'), [u'1', u'2', u'3', u'4'])
 
     def test_non_default_encoding(self):
         """#13572 - QueryDict with a non-default encoding"""
-        q = QueryDict('sbb=one', encoding='rot_13') 
+        q = QueryDict('sbb=one', encoding='rot_13')
         self.assertEqual(q.encoding , 'rot_13' )
         self.assertEqual(q.items() , [(u'foo', u'bar')] )
         self.assertEqual(q.urlencode() , 'sbb=one' )
-        q = q.copy() 
+        q = q.copy()
         self.assertEqual(q.encoding , 'rot_13' )
         self.assertEqual(q.items() , [(u'foo', u'bar')] )
         self.assertEqual(q.urlencode() , 'sbb=one' )
         self.assertEqual(copy.copy(q).encoding , 'rot_13' )
         self.assertEqual(copy.deepcopy(q).encoding , 'rot_13')
-        
+
 class HttpResponseTests(unittest.TestCase):
     def test_unicode_headers(self):
         r = HttpResponse()
@@ -203 +204 @@
         # If we insert a unicode value it will be converted to an ascii
         r['value'] = u'test value'
         self.failUnless(isinstance(r['value'], str))
-        
+
         # An error is raised When a unicode object with non-ascii is assigned.
         self.assertRaises(UnicodeEncodeError, r.__setitem__, 'value', u't\xebst value')
-        
-        # The response also converts unicode keys to strings.)      
+
+        # The response also converts unicode keys to strings.)
         r[u'test'] = 'testing key'
         l = list(r.items())
         l.sort()
         self.assertEqual(l[1], ('test', 'testing key'))
-        
+
         # It will also raise errors for keys with non-ascii data.
         self.assertRaises(UnicodeEncodeError, r.__setitem__, u't\xebst key', 'value')
 
@@ -254 +255 @@
         c2 = CompatCookie()
         c2.load(c.output())
         self.assertEqual(c['test'].value, c2['test'].value)
+
+class HttpResponseRedirectTests(unittest.TestCase):
+    def test_open_redirect(self):
+        self.assertEqual('/test', HttpResponseRedirect('/test')['Location'])
+        self.assertEqual(settings.LOGIN_REDIRECT_URL, HttpResponseRedirect('http://djangoproject.com')['Location'])
+
+        r = HttpResponseRedirect('http://djangoproject.com', whitelist=[r'.*'])
+        self.assertEqual('http://djangoproject.com', r['Location'])
+
+        setattr(settings, 'REDIRECT_WHITELIST', [r'.*'])
+        self.assertEqual('http://djangoproject.com', HttpResponseRedirect('http://djangoproject.com')['Location'])