security: upgrade auth hashing and request signatures

cmdb-api/api/lib/http_cli.py

@@ -17,7 +17,7 @@ def build_api_key(path, params):
     values = "".join([str(params[k]) for k in sorted(params.keys())
                       if params[k] is not None]) if params.keys() else ""
     _secret = "".join([path, secret, values]).encode("utf-8")
-    params["_secret"] = hashlib.sha1(_secret).hexdigest()
+    params["_secret"] = hashlib.sha256(_secret).hexdigest()
     params["_key"] = key
 
     return params

cmdb-api/api/lib/perm/acl/acl.py

@@ -31,7 +31,7 @@
 def get_access_token():
     url = "{0}/acl/apps/token".format(current_app.config.get('ACL_URI'))
     payload = dict(app_id=current_app.config.get('APP_ID'),
-                   secret_key=hashlib.md5(current_app.config.get('APP_SECRET_KEY').encode('utf-8')).hexdigest())
+                   secret_key=hashlib.sha256(current_app.config.get('APP_SECRET_KEY').encode('utf-8')).hexdigest())
     try:
         res = requests.post(url, data=payload).json()
         return res.get("token")

cmdb-api/api/lib/perm/acl/app.py

@@ -2,6 +2,7 @@
 
 import datetime
 import hashlib
+import hmac
 
 import jwt
 from flask import abort
@@ -79,7 +80,12 @@ def _get_by_key(key):
     @classmethod
     def gen_token(cls, key, secret):
         app = cls._get_by_key(key) or abort(404, ErrFormat.app_not_found.format("key={}".format(key)))
-        secret != hashlib.md5(app.secret_key.encode('utf-8')).hexdigest() and abort(403, ErrFormat.app_secret_invalid)
+        if not isinstance(secret, str):
+            abort(403, ErrFormat.app_secret_invalid)
+        secret_sha256 = hashlib.sha256(app.secret_key.encode('utf-8')).hexdigest()
+        secret_md5 = hashlib.md5(app.secret_key.encode('utf-8')).hexdigest()
+        if not (hmac.compare_digest(secret_sha256, secret) or hmac.compare_digest(secret_md5, secret)):
+            abort(403, ErrFormat.app_secret_invalid)
 
         token = jwt.encode({
             'sub': app.name,

cmdb-api/api/models/acl.py

@@ -3,12 +3,14 @@
 
 import copy
 import hashlib
+import hmac
 from datetime import datetime
 
 from flask import current_app
 from flask import session
 from flask_sqlalchemy import BaseQuery
 
+from api.extensions import bcrypt
 from api.extensions import db
 from api.lib.database import CRUDModel
 from api.lib.database import Model
@@ -19,6 +21,32 @@
 from api.lib.perm.acl.resp_format import ErrFormat
 
 
+def _build_signatures(path, secret, args):
+    values = "".join([str(i) for i in (args or [])])
+    payload = '{0}{1}{2}'.format(path or "", secret or "", values).encode("utf-8")
+
+    return {
+        "sha256": hashlib.sha256(payload).hexdigest(),
+        "sha1": hashlib.sha1(payload).hexdigest(),
+    }
+
+
+def _verify_signature(path, secret, args, provided):
+    if isinstance(provided, bytes):
+        provided = provided.decode("utf-8", "ignore")
+    if not isinstance(provided, str):
+        return False
+
+    signatures = _build_signatures(path, secret, args)
+
+    return (hmac.compare_digest(signatures["sha256"], provided) or
+            hmac.compare_digest(signatures["sha1"], provided))
+
+
+def _is_bcrypt_hash(password):
+    return isinstance(password, str) and password.startswith(("$2a$", "$2b$", "$2y$"))
+
+
 class App(Model):
     __tablename__ = "acl_apps"
 
@@ -55,8 +83,7 @@ def authenticate_with_key(self, key, secret, args, path):
         user = self.filter(User.key == key).filter(User.deleted.is_(False)).filter(User.block == 0).first()
         if not user:
             return None, False
-        if user and hashlib.sha1('{0}{1}{2}'.format(
-                path, user.secret, "".join(args)).encode("utf-8")).hexdigest() == secret:
+        if user and _verify_signature(path, user.secret, args, secret):
             authenticated = True
         else:
             authenticated = False
@@ -132,14 +159,21 @@ def _get_password(self):
         return self._password
 
     def _set_password(self, password):
-        self._password = hashlib.md5(password.encode('utf-8')).hexdigest()
+        if password:
+            self._password = bcrypt.generate_password_hash(password).decode('utf-8')
 
     password = db.synonym("_password", descriptor=property(_get_password, _set_password))
 
     def check_password(self, password):
         if self.password is None:
             return False
-        return self.password == password or self.password == hashlib.md5(password.encode('utf-8')).hexdigest()
+        if _is_bcrypt_hash(self.password):
+            try:
+                return bcrypt.check_password_hash(self.password, password)
+            except ValueError:
+                return False
+        legacy_md5 = hashlib.md5(password.encode('utf-8')).hexdigest()
+        return self.password == password or self.password == legacy_md5
 
 
 class RoleQuery(BaseQuery):
@@ -162,8 +196,7 @@ def authenticate_with_key(self, key, secret, args, path):
         role = self.filter(Role.key == key).filter(Role.deleted.is_(False)).first()
         if not role:
             return None, False
-        if role and hashlib.sha1('{0}{1}{2}'.format(
-                path, role.secret, "".join(args)).encode("utf-8")).hexdigest() == secret:
+        if role and _verify_signature(path, role.secret, args, secret):
             authenticated = True
         else:
             authenticated = False
@@ -188,14 +221,20 @@ def _get_password(self):
 
     def _set_password(self, password):
         if password:
-            self._password = hashlib.md5(password.encode('utf-8')).hexdigest()
+            self._password = bcrypt.generate_password_hash(password).decode('utf-8')
 
     password = db.synonym("_password", descriptor=property(_get_password, _set_password))
 
     def check_password(self, password):
         if self.password is None:
             return False
-        return self.password == password or self.password == hashlib.md5(password.encode('utf-8')).hexdigest()
+        if _is_bcrypt_hash(self.password):
+            try:
+                return bcrypt.check_password_hash(self.password, password)
+            except ValueError:
+                return False
+        legacy_md5 = hashlib.md5(password.encode('utf-8')).hexdigest()
+        return self.password == password or self.password == legacy_md5
 
 
 class RoleRelation(Model):