feat: store bcrypt encoded passwords

requirements.txt

@@ -8,3 +8,4 @@ python-ldap==3.3.1
 starlette==0.14.2
 typing-extensions==3.10.0.0
 uvicorn==0.13.4
+bcrypt==3.2.0

src/main.py

@@ -1,31 +1,40 @@
+import bcrypt
 import ldap
-
 from fastapi import FastAPI, HTTPException, Response
 from pydantic import BaseModel
-from ldap import modlist
 
-from config import LDAP_URI, LDAP_BASE_DN
+from config import LDAP_BASE_DN, LDAP_URI
 
 app = FastAPI()
 
+
 class PasswordUpdate(BaseModel):
-  bind_pw: str
+    bind_pw: str
-  userPassword: str
+    userPassword: str
+
 
 @app.post("/users/{rdn}/updatePassword", status_code=204, response_class=Response)
-def change_password(rdn: str, updateRequest: PasswordUpdate):
+def change_password(rdn: str, update_request: PasswordUpdate):
-  try:
+    try:
-    ldap_conn = _connect_ldap_simple_bind(LDAP_URI, f"{rdn},{LDAP_BASE_DN}", updateRequest.bind_pw)
+        ldap_conn = _connect_ldap_simple_bind(LDAP_URI, f"{rdn},{LDAP_BASE_DN}", update_request.bind_pw)
-  except ldap.INVALID_CREDENTIALS as e:
+    except ldap.INVALID_CREDENTIALS as e:
-    raise HTTPException(status_code=401, detail=str(e))
+        raise HTTPException(status_code=401, detail=str(e))
-  _update_ldap_userPassword(ldap_conn, f"{rdn},{LDAP_BASE_DN}", updateRequest.userPassword)
+    new_pass = _hash_password(update_request.userPassword)
+    _update_ldap_userPassword(ldap_conn, f"{rdn},{LDAP_BASE_DN}", new_pass)
+
 
 def _connect_ldap_simple_bind(server_uri: str, bind_dn: str, bind_pw: str):
-  ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
+    ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
-  conn = ldap.initialize(server_uri)
+    conn = ldap.initialize(server_uri)
-  conn.simple_bind_s(bind_dn, bind_pw)
+    conn.simple_bind_s(bind_dn, bind_pw)
-  return conn
+    return conn
+
 
 def _update_ldap_userPassword(conn, dn: str, new_pass: str):
-  changes = [( ldap.MOD_REPLACE, 'userPassword', bytes(str(new_pass), 'utf-8') )]
+    changes = [( ldap.MOD_REPLACE, 'userPassword', bytes(str(new_pass), 'utf-8') )]
-  result = conn.modify_ext_s(dn, changes)
+    result = conn.modify_ext_s(dn, changes)
+
+
+def _hash_password(pw: str):
+    hash_b = bcrypt.hashpw(pw.encode(), bcrypt.gensalt())
+    return '{BCRYPT}' + hash_b.decode()