Fix for ticket tornadoweb#904 - Addition of get_claimed_ip() to HTTPR…

…equest

As per ticket description of ticket tornadoweb#904, a method to return the first
public IP address from X-Forwarded-For should be implemented. This
update contains the below changes:

1. In tornado/httpserver.py
HTTPRequest.get_claimed_ip() is implemented which simply returns
remote_ip

2. In tornado/test/httpserver_test.py
The test for verifying ip_headers already existed in XHeaders. I simply
modified the internal class Handler to retrieve the output of
get_claimed_ip from the request messge and write it to the dictionary.
The same tests that were written for verifying ip_headers have been
re-used except that they now check that remote ip equals claimed_ip

tornado/httpserver.py

@@ -542,6 +542,10 @@ def get_ssl_certificate(self, binary_form=False):
         except ssl.SSLError:
             return None
 
+    def get_claimed_ip(self):
+        """Returns the first public IP address from the X-Forwarded-For"""
+        return self.remote_ip
+
     def __repr__(self):
         attrs = ("protocol", "host", "method", "uri", "version", "remote_ip")
         args = ", ".join(["%s=%r" % (n, getattr(self, n)) for n in attrs])

tornado/test/httpserver_test.py

@@ -402,7 +402,8 @@ class XHeaderTest(HandlerBaseTestCase):
     class Handler(RequestHandler):
         def get(self):
             self.write(dict(remote_ip=self.request.remote_ip,
-                            remote_protocol=self.request.protocol))
+                            remote_protocol=self.request.protocol,
+                            claimed_ip=self.request.get_claimed_ip()))
 
     def get_httpserver_options(self):
         return dict(xheaders=True)
@@ -462,8 +463,46 @@ def test_scheme_headers(self):
         self.assertEqual(
             self.fetch_json("/", headers=bad_forwarded)["remote_protocol"],
             "http")
+
+    def test_get_claimed_ip(self):
+        self.assertEqual(self.fetch_json("/")["remote_ip"], 
+                         self.fetch_json("/")["claimed_ip"])
 
+        valid_ipv4 = {"X-Real-IP": "4.4.4.4"}
+        self.assertEqual(
+            self.fetch_json("/", headers=valid_ipv4)["remote_ip"],
+            self.fetch_json("/", headers=valid_ipv4)["claimed_ip"])
+
+        valid_ipv4_list = {"X-Forwarded-For": "127.0.0.1, 4.4.4.4"}
+        self.assertEqual(
+            self.fetch_json("/", headers=valid_ipv4_list)["remote_ip"],
+            self.fetch_json("/", headers=valid_ipv4_list)["claimed_ip"])
 
+        valid_ipv6 = {"X-Real-IP": "2620:0:1cfe:face:b00c::3"}
+        self.assertEqual(
+            self.fetch_json("/", headers=valid_ipv6)["remote_ip"],
+            self.fetch_json("/", headers=valid_ipv6)["claimed_ip"])
+
+        valid_ipv6_list = {"X-Forwarded-For": "::1, 2620:0:1cfe:face:b00c::3"}
+        self.assertEqual(
+            self.fetch_json("/", headers=valid_ipv6_list)["remote_ip"],
+            self.fetch_json("/", headers=valid_ipv6_list)["claimed_ip"])
+
+        invalid_chars = {"X-Real-IP": "4.4.4.4<script>"}
+        self.assertEqual(
+            self.fetch_json("/", headers=invalid_chars)["remote_ip"],
+            "127.0.0.1")
+
+        invalid_chars_list = {"X-Forwarded-For": "4.4.4.4, 5.5.5.5<script>"}
+        self.assertEqual(
+            self.fetch_json("/", headers=invalid_chars_list)["remote_ip"],
+            self.fetch_json("/", headers=invalid_chars_list)["claimed_ip"])
+
+        invalid_host = {"X-Real-IP": "www.google.com"}
+        self.assertEqual(
+            self.fetch_json("/", headers=invalid_host)["remote_ip"],
+            self.fetch_json("/", headers=invalid_host)["claimed_ip"])
+
 class SSLXHeaderTest(AsyncHTTPSTestCase, HandlerBaseTestCase):
     def get_app(self):
         return Application([('/', XHeaderTest.Handler)])