7x31
/
ad-password-self-service
mirror of https://github.com/capricornxl/ad-password-self-service.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

BUG FIX

This commit is contained in:
向乐🌌 2021-05-22 17:27:42 +08:00
parent 870cc106ba
commit 72fe535f3c
3 changed files with 173 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
resetpwd/utils.py
View File

@ -10,6 +10,7 @@ from django.shortcuts import render
from django.http import HttpResponseRedirect from django.http import HttpResponseRedirect
import logging import logging
from utils.crypto_ops import Crypto from utils.crypto_ops import Crypto
from ldap3.core.exceptions import LDAPInvalidCredentialsResult, LDAPOperationResult, LDAPExceptionError, LDAPException
from django.conf import settings from django.conf import settings
from pwdselfservice import crypto_key from pwdselfservice import crypto_key
@ -23,7 +24,10 @@ else:
logger = logging.getLogger('django') logger = logging.getLogger('django')
def code_2_user_id(ops, request, msg_template, home_url, code): def code_2_user_info(ops, request, msg_template, home_url, code):
"""
临时授权码换取userinfo
"""
_status, user_id = ops.get_user_id_by_code(code) _status, user_id = ops.get_user_id_by_code(code)
# 判断 user_id 在本企业钉钉/微信中是否存在 # 判断 user_id 在本企业钉钉/微信中是否存在
if not _status: if not _status:
@ -45,6 +49,9 @@ def code_2_user_id(ops, request, msg_template, home_url, code):
def crypto_id_2_user_info(ops, request, msg_template, home_url, scan_app_tag): def crypto_id_2_user_info(ops, request, msg_template, home_url, scan_app_tag):
"""
能过前端提交的加密的userid来获取用户信息<userinfo>
"""
try: try:
crypto_tmp_id = request.COOKIES.get('tmpid') crypto_tmp_id = request.COOKIES.get('tmpid')
if not crypto_tmp_id: if not crypto_tmp_id:
@ -73,6 +80,9 @@ def crypto_id_2_user_info(ops, request, msg_template, home_url, scan_app_tag):
def crypto_user_id_2_cookie(user_id): def crypto_user_id_2_cookie(user_id):
"""
加密userid，写入到cookie
"""
crypto = Crypto(crypto_key) crypto = Crypto(crypto_key)
# 对user_id进行加密，因为user_id基本上固定不变的，为了防止user_id泄露而导致重复使用，进行加密后再传回。 # 对user_id进行加密，因为user_id基本上固定不变的，为了防止user_id泄露而导致重复使用，进行加密后再传回。
_id_cryto = crypto.encrypt(user_id) _id_cryto = crypto.encrypt(user_id)
@ -83,6 +93,9 @@ def crypto_user_id_2_cookie(user_id):
def crypto_id_2_user_id(request, msg_template, home_url): def crypto_id_2_user_id(request, msg_template, home_url):
"""
前端提交的加密的userid，解密出真实的userid
"""
try: try:
crypto_tmp_id = request.COOKIES.get('tmpid') crypto_tmp_id = request.COOKIES.get('tmpid')
# 解密 # 解密
@ -100,55 +113,74 @@ def crypto_id_2_user_id(request, msg_template, home_url):
def ops_account(ad_ops, request, msg_template, home_url, username, new_password): def ops_account(ad_ops, request, msg_template, home_url, username, new_password):
if ad_ops.ad_ensure_user_by_account(username) is False: """
context = { ad 账号操作，判断账号状态，重置密码或解锁账号
'msg': "账号[%s]在AD中不存在，请确认当前钉钉扫码账号绑定的邮箱是否和您正在使用的邮箱一致？或者该账号己被禁用！\n猜测：您的账号或邮箱是否是带有数字或其它字母区分？" % username, """
'button_click': "window.location.href='%s'" % home_url, try:
'button_display': "返回主页" print("ops_account: {}" .format(username))
} _status, _account = ad_ops.ad_ensure_user_by_account(username=username)
return render(request, msg_template, context) if not _status:
context = {
'msg': "账号[%s]在AD中不存在，请确认当前钉钉扫码账号绑定的邮箱是否和您正在使用的邮箱一致？或者该账号己被禁用！\n猜测：您的账号或邮箱是否是带有数字或其它字母区分？" % username,
'button_click': "window.location.href='%s'" % home_url,
'button_display': "返回主页"
}
return render(request, msg_template, context)
account_code = ad_ops.ad_get_user_status_by_account(username) _status, account_code = ad_ops.ad_get_user_status_by_account(username)
if account_code in settings.AD_ACCOUNT_DISABLE_CODE: if _status and account_code in settings.AD_ACCOUNT_DISABLE_CODE:
context = { context = {
'msg': "此账号状态为己禁用，请联系HR确认账号是否正确。", 'msg': "此账号状态为己禁用，请联系HR确认账号是否正确。",
'button_click': "window.location.href='%s'" % home_url, 'button_click': "window.location.href='%s'" % home_url,
'button_display': "返回主页" 'button_display': "返回主页"
} }
return render(request, msg_template, context) return render(request, msg_template, context)
elif not _status:
context = {
'msg': "错误：{}" .format(account_code),
'button_click': "window.location.href='%s'" % home_url,
'button_display': "返回主页"
}
return render(request, msg_template, context)
if new_password: if new_password:
reset_status, result = ad_ops.ad_reset_user_pwd_by_account(username=username, new_password=new_password) reset_status, result = ad_ops.ad_reset_user_pwd_by_account(username=username, new_password=new_password)
if reset_status: if reset_status:
# 重置密码并执行一次解锁，防止重置后账号还是锁定状态。 # 重置密码并执行一次解锁，防止重置后账号还是锁定状态。
unlock_status, result = ad_ops.ad_unlock_user_by_account(username) unlock_status, result = ad_ops.ad_unlock_user_by_account(username)
if unlock_status: if unlock_status:
context = {
'msg': "密码己修改/重置成功，请妥善保管。你可以点击返回主页或直接关闭此页面！",
'button_click': "window.location.href='%s'" % home_url,
'button_display': "返回主页"
}
return render(request, msg_template, context)
else:
context = { context = {
'msg': "密码己修改/重置成功，请妥善保管。你可以点击返回主页或直接关闭此页面！", 'msg': "密码未修改/重置成功，错误信息：{}".format(result),
'button_click': "window.location.href='%s'" % home_url, 'button_click': "window.location.href='%s'" % home_url,
'button_display': "返回主页" 'button_display': "返回主页"
} }
return render(request, msg_template, context) return render(request, msg_template, context)
else: else:
context = { unlock_status, result = ad_ops.ad_unlock_user_by_account(username)
'msg': "密码未修改/重置成功，错误信息：{}".format(result), if unlock_status:
'button_click': "window.location.href='%s'" % home_url, context = {
'button_display': "返回主页" 'msg': "账号己解锁成功。你可以点击返回主页或直接关闭此页面！",
} 'button_click': "window.location.href='%s'" % home_url,
return render(request, msg_template, context) 'button_display': "返回主页"
else: }
unlock_status, result = ad_ops.ad_unlock_user_by_account(username) return render(request, msg_template, context)
if unlock_status: else:
context = { context = {
'msg': "账号己解锁成功。你可以点击返回主页或直接关闭此页面！", 'msg': "账号未能解锁，错误信息：{}".format(result),
'button_click': "window.location.href='%s'" % home_url, 'button_click': "window.location.href='%s'" % home_url,
'button_display': "返回主页" 'button_display': "返回主页"
} }
return render(request, msg_template, context) return render(request, msg_template, context)
else: except LDAPInvalidCredentialsResult as lic_e:
context = { raise LDAPOperationResult("LDAPInvalidCredentialsResult: " + str(lic_e.message))
'msg': "账号未能解锁，错误信息：{}".format(result), except LDAPOperationResult as lo_e:
'button_click': "window.location.href='%s'" % home_url, raise LDAPOperationResult("LDAPOperationResult: " + str(lo_e.message))
'button_display': "返回主页" except LDAPException as l_e:
} raise LDAPException("LDAPException: " + str(l_e))
return render(request, msg_template, context)

39
resetpwd/views.py
View File

@ -2,9 +2,10 @@ import logging
import os import os
from django.shortcuts import render from django.shortcuts import render
from utils.ad_ops import AdOps from utils.ad_ops import AdOps
from ldap3.core.exceptions import LDAPException
from utils.format_username import format2username, get_user_is_active from utils.format_username import format2username, get_user_is_active
from .form import CheckForm from .form import CheckForm
from .utils import code_2_user_id, crypto_id_2_user_info, ops_account, crypto_id_2_user_id, crypto_user_id_2_cookie from .utils import code_2_user_info, crypto_id_2_user_info, ops_account, crypto_id_2_user_id, crypto_user_id_2_cookie
APP_ENV = os.getenv('APP_ENV') APP_ENV = os.getenv('APP_ENV')
if APP_ENV == 'dev': if APP_ENV == 'dev':
from conf.local_settings_dev import SCAN_CODE_TYPE, DING_MO_APP_ID, WEWORK_CORP_ID, WEWORK_AGENT_ID, HOME_URL from conf.local_settings_dev import SCAN_CODE_TYPE, DING_MO_APP_ID, WEWORK_CORP_ID, WEWORK_AGENT_ID, HOME_URL
@ -39,13 +40,13 @@ class PARAMS(object):
scan_params = PARAMS() scan_params = PARAMS()
_ops = scan_params.ops _ops = scan_params.ops
try: # try:
ad_ops = AdOps() # AdOps() = AdOps()
print("初始化Active Directory连接成功...") # print("初始化Active Directory连接成功...")
except Exception as e: # except Exception as e:
ad_ops = None # AdOps() = LDAPException("连接域控制器失败，无法访问到LDAP")
print("初始化Active Directory连接失败...") # print("初始化Active Directory连接失败...")
print(str(e)) # print(str(e))
def index(request): def index(request):
@ -86,7 +87,7 @@ def index(request):
# 格式化用户名 # 格式化用户名
username = format2username(username) username = format2username(username)
# 检测账号状态 # 检测账号状态
auth_status, auth_result = ad_ops.ad_auth_user(username=username, password=old_password) auth_status, auth_result = AdOps().ad_auth_user(username=username, password=old_password)
if not auth_status: if not auth_status:
context = { context = {
'msg': str(auth_result), 'msg': str(auth_result),
@ -94,7 +95,7 @@ def index(request):
'button_display': "返回" 'button_display': "返回"
} }
return render(request, msg_template, context) return render(request, msg_template, context)
return ops_account(ad_ops, request, msg_template, home_url, username, new_password) return ops_account(AdOps(), request, msg_template, home_url, username, new_password)
else: else:
context = { context = {
'msg': "请从主页进行修改密码操作或扫码验证用户信息。", 'msg': "请从主页进行修改密码操作或扫码验证用户信息。",
@ -124,7 +125,7 @@ def callback_check(request):
return render(request, msg_template, context) return render(request, msg_template, context)
print("code: {}" .format(code)) print("code: {}" .format(code))
try: try:
_status, user_id, user_info = code_2_user_id(_ops, request, msg_template, home_url, code) _status, user_id, user_info = code_2_user_info(_ops, request, msg_template, home_url, code)
if not _status: if not _status:
return render(request, msg_template, user_id) return render(request, msg_template, user_id)
# 账号是否是激活的 # 账号是否是激活的
@ -156,17 +157,9 @@ def reset_pwd_by_callback(request):
home_url = '%s://%s' % (request.scheme, HOME_URL) home_url = '%s://%s' % (request.scheme, HOME_URL)
# 从cookie中提取union_id，并解密，然后对当前union_id的用户进行重置密码 # 从cookie中提取union_id，并解密，然后对当前union_id的用户进行重置密码
if request.method == 'GET': if request.method == 'GET':
_status, user_id = crypto_id_2_user_id(request, msg_template, home_url) _status, user_info = crypto_id_2_user_info(_ops, request, msg_template, home_url, scan_params.SCAN_APP)
if not _status: if not _status:
return render(request, msg_template, user_id) return render(request, msg_template, user_info)
userid_status, user_info = _ops.get_user_detail_by_user_id(user_id)
if not userid_status:
context = {
'msg': '获取{}用户信息失败，错误信息：{}'.format(user_info, scan_params.SCAN_APP),
'button_click': "window.location.href='%s'" % home_url,
'button_display': "返回主页"
}
return render(request, msg_template, context)
# 通过user_id拿到用户信息，并格式化为username # 通过user_id拿到用户信息，并格式化为username
username = format2username(user_info.get('email')) username = format2username(user_info.get('email'))
# 如果邮箱能提取到，则格式化之后，提取出账号提交到前端绑定 # 如果邮箱能提取到，则格式化之后，提取出账号提交到前端绑定
@ -191,7 +184,7 @@ def reset_pwd_by_callback(request):
if not _status: if not _status:
return render(request, msg_template, user_info) return render(request, msg_template, user_info)
username = format2username(user_info.get('email')) username = format2username(user_info.get('email'))
return ops_account(ad_ops, request, msg_template, home_url, username, _new_password) return ops_account(ad_ops=AdOps(), request=request, msg_template=msg_template, home_url=home_url, username=username, new_password=_new_password)
except Exception as reset_e: except Exception as reset_e:
context = { context = {
'msg': "错误[%s]，请与管理员联系." % str(reset_e), 'msg': "错误[%s]，请与管理员联系." % str(reset_e),
@ -231,7 +224,7 @@ def unlock_account(request):
if not _status: if not _status:
return render(request, msg_template, user_info) return render(request, msg_template, user_info)
username = format2username(user_info.get('email')) username = format2username(user_info.get('email'))
return ops_account(ad_ops, request, msg_template, home_url, username, None) return ops_account(AdOps(), request, msg_template, home_url, username, None)
else: else:
context = { context = {
'msg': "请从主页开始进行操作。", 'msg': "请从主页开始进行操作。",

141
utils/ad_ops.py
View File

@ -1,5 +1,6 @@
import ldap3
from ldap3 import * from ldap3 import *
from ldap3.core.exceptions import LDAPInvalidCredentialsResult, LDAPOperationResult from ldap3.core.exceptions import LDAPInvalidCredentialsResult, LDAPOperationResult, LDAPExceptionError,LDAPException
from ldap3.core.results import * from ldap3.core.results import *
from ldap3.utils.dn import safe_dn from ldap3.utils.dn import safe_dn
import os import os
@ -54,10 +55,12 @@ class AdOps(object):
try: try:
self.conn = Connection(server, auto_bind=self.auto_bind, user=r'{}\{}'.format(self.domain, self.user), password=self.password, self.conn = Connection(server, auto_bind=self.auto_bind, user=r'{}\{}'.format(self.domain, self.user), password=self.password,
authentication=self.authentication, raise_exceptions=True) authentication=self.authentication, raise_exceptions=True)
except LDAPOperationResult as e: except LDAPInvalidCredentialsResult as lic_e:
raise LDAPOperationResult("LDAPOperationResult: " + str(e)) raise LDAPOperationResult("LDAPInvalidCredentialsResult: " + str(lic_e.message))
except Exception as e: except LDAPOperationResult as lo_e:
raise Exception('LDAP Exception：无法连接到AD控制器 --- {}' .format(e)) raise LDAPOperationResult("LDAPOperationResult: " + str(lo_e.message))
except LDAPException as l_e:
raise LDAPException("LDAPException: " + str(l_e))
def ad_auth_user(self, username, password): def ad_auth_user(self, username, password):
""" """
@ -100,10 +103,10 @@ class AdOps(object):
:param username: :param username:
:return: True or False :return: True or False
""" """
base_dn = BASE_DN try:
condition = '(&(objectclass=user)(sAMAccountName={}))'.format(username) return True, self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['sAMAccountName'])
attributes = ['sAMAccountName'] except Exception as e:
return self.conn.search(base_dn, condition, attributes=attributes) return False, "AdOps Exception: {}" .format(e)
def ad_get_user_displayname_by_account(self, username): def ad_get_user_displayname_by_account(self, username):
""" """
@ -111,11 +114,11 @@ class AdOps(object):
:param username: :param username:
:return: user_displayname :return: user_displayname
""" """
self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['name'])
try: try:
self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['name'])
return True, self.conn.entries[0]['name'] return True, self.conn.entries[0]['name']
except Exception as e: except Exception as e:
return False, str(e) return False, "AdOps Exception: {}" .format(e)
def ad_get_user_dn_by_account(self, username): def ad_get_user_dn_by_account(self, username):
""" """
@ -123,8 +126,11 @@ class AdOps(object):
:param username: :param username:
:return: DN :return: DN
""" """
self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['distinguishedName']) try:
return str(self.conn.entries[0]['distinguishedName']) self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['distinguishedName'])
return True, str(self.conn.entries[0]['distinguishedName'])
except Exception as e:
return False, "AdOps Exception: {}" .format(e)
def ad_get_user_status_by_account(self, username): def ad_get_user_status_by_account(self, username):
""" """
@ -132,8 +138,11 @@ class AdOps(object):
:param username: :param username:
:return: user_account_control code :return: user_account_control code
""" """
self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['userAccountControl']) try:
return self.conn.entries[0]['userAccountControl'] self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['userAccountControl'])
return True, self.conn.entries[0]['userAccountControl']
except Exception as e:
return False, "AdOps Exception: {}" .format(e)
def ad_unlock_user_by_account(self, username): def ad_unlock_user_by_account(self, username):
""" """
@ -141,11 +150,14 @@ class AdOps(object):
:param username: :param username:
:return: :return:
""" """
user_dn = self.ad_get_user_dn_by_account(username) _status, user_dn = self.ad_get_user_dn_by_account(username)
try: if _status:
return True, self.conn.extend.microsoft.unlock_account(user='%s' % user_dn) try:
except Exception as e: return True, self.conn.extend.microsoft.unlock_account(user='%s' % user_dn)
return False, str(e) except Exception as e:
return False, "AdOps Exception: {}".format(e)
else:
return False, user_dn
def ad_reset_user_pwd_by_account(self, username, new_password): def ad_reset_user_pwd_by_account(self, username, new_password):
""" """
@ -153,33 +165,35 @@ class AdOps(object):
:param username: :param username:
:return: :return:
""" """
user_dn = self.ad_get_user_dn_by_account(username) _status, user_dn = self.ad_get_user_dn_by_account(username)
if self.conn.check_names: if _status:
user_dn = safe_dn(user_dn) if self.conn.check_names:
encoded_new_password = ('"%s"' % new_password).encode('utf-16-le') user_dn = safe_dn(user_dn)
result = self.conn.modify(user_dn, encoded_new_password = ('"%s"' % new_password).encode('utf-16-le')
{'unicodePwd': [(MODIFY_REPLACE, [encoded_new_password])]}, result = self.conn.modify(user_dn,
) {'unicodePwd': [(MODIFY_REPLACE, [encoded_new_password])]},
)
if not self.conn.strategy.sync: if not self.conn.strategy.sync:
_, result = self.conn.get_response(result) _, result = self.conn.get_response(result)
else:
if self.conn.strategy.thread_safe:
_, result, _, _ = result
else: else:
result = self.conn.result if self.conn.strategy.thread_safe:
_, result, _, _ = result
else:
result = self.conn.result
# change successful, returns True # change successful, returns True
if result['result'] == RESULT_SUCCESS: if result['result'] == RESULT_SUCCESS:
return True, '密码己修改成功，请妥善保管！' return True, '密码己修改成功，请妥善保管！'
# change was not successful, raises exception if raise_exception = True in connection or returns the operation result, error code is in result['result'] # change was not successful, raises exception if raise_exception = True in connection or returns the operation result, error code is in result['result']
if self.conn.raise_exceptions: if self.conn.raise_exceptions:
from ldap3.core.exceptions import LDAPOperationResult from ldap3.core.exceptions import LDAPOperationResult
_msg = LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'], _msg = LDAPOperationResult(result=result['result'], description=result['description'], dn=result['dn'], message=result['message'],
response_type=result['type']) response_type=result['type'])
return False, _msg return False, _msg
return False, result['result'] return False, result['result']
else:
return False, user_dn
def ad_get_user_locked_status_by_account(self, username): def ad_get_user_locked_status_by_account(self, username):
""" """
@ -187,12 +201,15 @@ class AdOps(object):
:param username: :param username:
:return: 如果结果是1601-01-01说明账号未锁定，返回0 :return: 如果结果是1601-01-01说明账号未锁定，返回0
""" """
self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['lockoutTime']) try:
locked_status = self.conn.entries[0]['lockoutTime'] self.conn.search(BASE_DN, '(&(objectclass=user)(sAMAccountName={}))'.format(username), attributes=['lockoutTime'])
if '1601-01-01' in str(locked_status): locked_status = self.conn.entries[0]['lockoutTime']
return 0 if '1601-01-01' in str(locked_status):
else: return True, 'unlocked'
return locked_status else:
return False, locked_status
except Exception as e:
return False, "AdOps Exception: {}" .format(e)
if __name__ == '__main__': if __name__ == '__main__':
@ -204,14 +221,16 @@ if __name__ == '__main__':
# print(conn.result) # print(conn.result)
# conn = _ad_connect() # conn = _ad_connect()
user = 'zhangsan' # user = 'zhangsan'
old_password = 'K2dhhuT1Zf11111cnJ1ollC3y' # old_password = 'K2dhhuT1Zf11111cnJ1ollC3y'
# old_password = 'L1qyrmZDUFeYW1OIualjlNhr4' # # old_password = 'L1qyrmZDUFeYW1OIualjlNhr4'
new_password = 'K2dhhuT1Zf11111cnJ1ollC3y' # new_password = 'K2dhhuT1Zf11111cnJ1ollC3y'
ad_ops = AdOps() # ad_ops = AdOps()
# ad_ops = AdOps(user=user, password=old_password) # # ad_ops = AdOps(user=user, password=old_password)
status, msg = ad_ops.ad_auth_user(username=user, password=old_password) # status, msg = ad_ops.ad_auth_user(username=user, password=old_password)
print(msg) # print(msg)
if status: # if status:
res = ad_ops.ad_reset_user_pwd_by_account(user, new_password) # res = ad_ops.ad_reset_user_pwd_by_account(user, new_password)
print(res) # print(res)
_ad = AdOps()
print(_ad.ad_ensure_user_by_account('le.xiang'))