feat: first working draft

.gitignore

@@ -0,0 +1,2 @@
+env/
+src/__pycache__/

requirements.txt

@@ -0,0 +1,11 @@
+click             7.1.2
+fastapi           0.65.1
+h11               0.12.0
+pip               21.1.1
+pyasn1            0.4.8
+pydantic          1.8.2
+python3-ldap      0.9.8.4
+setuptools        49.2.1
+starlette         0.14.2
+typing-extensions 3.10.0.0
+uvicorn           0.13.4

src/main.py

@@ -1,9 +1,11 @@
 import ldap
 
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException, Response
 from pydantic import BaseModel
+from ldap import modlist
 
-LDAP_URI = "ldap://127.0.0.1:389"
+LDAP_URI = "ldap://10.42.0.1:389"
+LDAP_BASE_DN = "ou=users,dc=finallycoffee,dc=eu"
 
 app = FastAPI()
 
@@ -11,19 +13,21 @@ class PasswordUpdate(BaseModel):
   bind_pw: str
   userPassword: str
 
-@app.post("/users/{dn}/updatePassword")
-def change_password(dn: str, updateRequest: PasswordUpdate):
-  ldap_conn = _connect_ldap_simple_bind(LDAP_URI, dn, updateRequest.bind_pw)
-  _update_ldap_userPassword(ldap_conn, updateRequest.userPassword)
+@app.post("/users/{rdn}/updatePassword", status_code=204, response_class=Response)
+def change_password(rdn: str, updateRequest: PasswordUpdate):
+  try:
+    ldap_conn = _connect_ldap_simple_bind(LDAP_URI, f"{rdn},{LDAP_BASE_DN}", updateRequest.bind_pw)
+  except ldap.INVALID_CREDENTIALS as e:
+    raise HTTPException(status_code=401, detail=str(e))
+  _update_ldap_userPassword(ldap_conn, f"{rdn},{LDAP_BASE_DN}", updateRequest.userPassword)
 
-def _connect_ldap_simple_bind(server_uri: str, bind_dn: str, bind_pw: str) -> LDAPObject:
+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)
   conn = ldap.initialize(server_uri)
   conn.simple_bind_s(bind_dn, bind_pw)
   return conn
 
-def _update_ldap_userPassword(conn: LDAPObject, new_pass: str):
-  modlist = [( ldap.MOD_REPLACE, 'userPassword', new_pass )]
-  ldap.connection.modify_s(ldap.dn, modlist)
-
-
+def _update_ldap_userPassword(conn, dn: str, new_pass: str):
+  changes = [( ldap.MOD_REPLACE, 'userPassword', bytes(str(new_pass), 'utf-8') )]
+  result = conn.modify_ext_s(dn, changes)
+  print(changes, result)