fix(security): replace passwords whereever they could be used for output

Author: KaiSchwarz-cnic

hexonet/apiconnector/apiclient.py

@@ -90,16 +90,23 @@ def disableDebugMode(self):
         self.__debugMode = False
         return self
 
-    def getPOSTData(self, cmd):
+    def getPOSTData(self, cmd, secured=False):
         """
         Serialize given command for POST request including connection configuration data
         """
         data = self.__socketConfig.getPOSTData()
+        if secured:
+            data = re.sub(r's_pw=[^&]+', 's_pw=***', data)
         tmp = ""
         if not isinstance(cmd, str):
             for key in sorted(cmd.keys()):
                 if (cmd[key] is not None):
                     tmp += ("{0}={1}\n").format(key, re.sub('[\r\n]', '', str(cmd[key])))
+        else:
+            tmp = cmd
+        tmp = tmp.rstrip('\n')
+        if secured:
+            tmp = re.sub(r'PASSWORD=[^\n]+', 'PASSWORD=***', tmp)
         return ("{0}{1}={2}").format(data, quote('s_command'), quote(re.sub('\n$', '', tmp)))
 
     def getSession(self):
@@ -255,6 +262,7 @@ def request(self, cmd):
 
         # request command to API
         data = self.getPOSTData(newcmd).encode('UTF-8')
+        secured = self.getPOSTData(newcmd, True).encode('UTF-8')
         # TODO: 300s (to be sure to get an API response)
         try:
             headers = {
@@ -268,11 +276,11 @@ def request(self, cmd):
                 req.set_proxy(proxyurl.netloc, proxyurl.scheme)
             body = urlopen(req, timeout=self.__socketTimeout).read()
             if (self.__debugMode):
-                print((self.__socketURL, data, body, '\n', '\n'))
+                print((self.__socketURL, secured, body, '\n', '\n'))
         except Exception:
             body = rtm.getTemplate("httperror").getPlain()
             if (self.__debugMode):
-                print((self.__socketURL, data, "HTTP communication failed", body, '\n', '\n'))
+                print((self.__socketURL, secured, "HTTP communication failed", body, '\n', '\n'))
         return Response(body, newcmd)
 
     def requestNextResponsePage(self, rr):

hexonet/apiconnector/response.py

@@ -24,6 +24,8 @@ def __init__(self, raw, cmd=None):
         super(Response, self).__init__(raw)
         # The API Command used within this request
         self.__command = cmd
+        if (self.__command is not None) and ('PASSWORD' in self.__command):
+            self.__command['PASSWORD'] = '***'
         # Column names available in this responsse.
         # NOTE: this includes also FIRST, LAST, LIMIT, COUNT, TOTAL
         # and maybe further specific columns in case of a list query

tests/test_apiclient.py

@@ -79,7 +79,7 @@ def test_apiclientmethods():
 
     # test string input
     enc = cl.getPOSTData('gregergege')
-    assert enc == 's_entity=54cd&s_command='
+    assert enc == 's_entity=54cd&s_command=gregergege'
 
     # test object input with null value in parameter
     validate = 's_entity=54cd&s_command=COMMAND%3DModifyDomain'
@@ -89,6 +89,20 @@ def test_apiclientmethods():
     })
     assert enc == validate
 
+    # test secured passwords
+    cl.setCredentials('test.user', 'test.passw0rd')
+    enc = cl.getPOSTData({
+        'COMMAND': 'CheckAuthentication',
+        'SUBUSER': 'test.user',
+        'PASSWORD': 'test.passw0rd'
+    }, True)
+    cl.setCredentials('', '')
+    expected = (
+        's_entity=54cd&s_login=test.user&s_pw=***&' +
+        's_command=COMMAND%3DCheckAuthentication%0APASSWORD%3D%2A%2A%2A%0ASUBUSER%3Dtest.user'
+    )
+    assert expected == enc
+
     # #.enableDebugMode()
     cl.enableDebugMode()
     cl.disableDebugMode()

tests/test_response.py

@@ -41,6 +41,7 @@ def test_responsemethods():
     assert r.getFirstRecordIndex() == 0
 
     # #.getCommandPlain()
+    # case 1
     r = R("", {
         "COMMAND": "QueryDomainOptions",
         "DOMAIN0": "example.com",
@@ -49,6 +50,15 @@ def test_responsemethods():
     expected = "COMMAND = QueryDomainOptions\nDOMAIN0 = example.com\nDOMAIN1 = example.net\n"
     assert r.getCommandPlain() == expected
 
+    # case secured
+    r = R("", {
+        "COMMAND": "CheckAuthentication",
+        "PASSWORD": "test.passw0rd",
+        "SUBUSER": "test.user"
+    })
+    expected = "COMMAND = CheckAuthentication\nPASSWORD = ***\nSUBUSER = test.user\n"
+    assert r.getCommandPlain() == expected
+
     # #.getColumns()
     r = R(rtm.getTemplate('listP0').getPlain())
     cols = r.getColumns()