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

fix some bugs

This commit is contained in:
Leven 2023-02-06 22:01:00 +08:00
parent a8afae7c2f
commit 1752b1d16f
6 changed files with 332 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
pwdselfservice/settings.py
View File

@ -1,4 +1,6 @@
import logging.config
import os import os
from django.utils.log import DEFAULT_LOGGING
APP_ENV = os.getenv('APP_ENV') APP_ENV = os.getenv('APP_ENV')
if APP_ENV == 'dev': if APP_ENV == 'dev':
@ -24,49 +26,49 @@ LOG_PATH = os.path.join(BASE_DIR, 'log')
if not os.path.isdir(LOG_PATH): if not os.path.isdir(LOG_PATH):
os.mkdir(LOG_PATH) os.mkdir(LOG_PATH)
LOGGING = { # Disable Django's logging setup
LOGGING_CONFIG = None
LOGLEVEL = os.environ.get('LOGLEVEL', 'info').upper()
logging.config.dictConfig({
'version': 1, 'version': 1,
# 此选项开启表示禁用部分日志，不建议设置为True
'disable_existing_loggers': False, 'disable_existing_loggers': False,
'formatters': { 'formatters': {
'verbose': { 'default': {
'format': '%(asctime)s %(levelname)s %(message)s' # exact format is not important, this is the minimum information
}, "format": "%(asctime)s %(module)s %(levelname)s -%(thread)d- %(message)s"
'simple': {
'format': '%(asctime)s %(levelname)s %(message)s'
},
},
'filters': {
'require_debug_true': {
# 过滤器，只有当setting的DEBUG = True时生效
'()': 'django.utils.log.RequireDebugTrue',
}, },
'django.server': DEFAULT_LOGGING['formatters']['django.server'],
}, },
'handlers': { 'handlers': {
# console logs to stderr
'console': { 'console': {
'level': 'DEBUG', "level": "DEBUG",
'filters': ['require_debug_true'],
'class': 'logging.StreamHandler', 'class': 'logging.StreamHandler',
'formatter': 'verbose' 'formatter': 'default',
}, },
'file': { 'file': {
'level': 'DEBUG', 'level': "DEBUG",
'class': 'logging.FileHandler', 'class': 'logging.handlers.RotatingFileHandler',
# 日志保存文件 'filename': os.path.join(LOG_PATH, "all.log"),
'filename': '%s/all.log' % LOG_PATH, 'formatter': 'default',
# 日志格式，与上边的设置对应选择 'maxBytes': 1024 * 1024 * 50, # 日志大小 10M
'formatter': 'verbose' 'backupCount': 24, # 备份数为 2# 简单格式
} 'encoding': 'utf-8',
},
'django.server': DEFAULT_LOGGING['handlers']['django.server'],
}, },
'loggers': { 'loggers': {
'django': { # default for all undefined Python modules
# 日志记录器 '': {
'level': LOGLEVEL,
'handlers': ['file'], 'handlers': ['file'],
'level': 'DEBUG', },
'propagate': True, # Default runserver request logging
} 'django.server': DEFAULT_LOGGING['loggers']['django.server'],
}, },
} })
# SESSION # SESSION

6
resetpwd/utils.py
View File

@ -11,6 +11,7 @@ import logging
from ldap3.core.exceptions import LDAPException from ldap3.core.exceptions import LDAPException
from django.conf import settings from django.conf import settings
import os import os
from utils.tracecalls import decorator_logger
APP_ENV = os.getenv('APP_ENV') APP_ENV = os.getenv('APP_ENV')
if APP_ENV == 'dev': if APP_ENV == 'dev':
@ -18,9 +19,10 @@ if APP_ENV == 'dev':
else: else:
from conf.local_settings import * from conf.local_settings import *
logger = logging.getLogger('django') logger = logging.getLogger(__name__)
@decorator_logger(logger, log_head='AccountOps', pretty=True, indent=2, verbose=1)
def code_2_user_detail(ops, home_url, code): def code_2_user_detail(ops, home_url, code):
""" """
临时授权码换取userinfo 临时授权码换取userinfo
@ -29,6 +31,7 @@ def code_2_user_detail(ops, home_url, code):
return _, s, e return _, s, e
@decorator_logger(logger, log_head='AccountOps', pretty=True, indent=2, verbose=1)
def code_2_user_info_with_oauth2(ops, request, msg_template, home_url, code): def code_2_user_info_with_oauth2(ops, request, msg_template, home_url, code):
""" """
临时授权码换取userinfo 临时授权码换取userinfo
@ -53,6 +56,7 @@ def code_2_user_info_with_oauth2(ops, request, msg_template, home_url, code):
return True, user_id, user_info return True, user_id, user_info
@decorator_logger(logger, log_head='AccountOps', pretty=True, indent=2, verbose=1)
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):
""" """
ad 账号操作，判断账号状态，重置密码或解锁账号 ad 账号操作，判断账号状态，重置密码或解锁账号

249
resetpwd/views.py
View File

@ -9,6 +9,7 @@ from .form import CheckForm
from .utils import code_2_user_detail, ops_account from .utils import code_2_user_detail, ops_account
from django.conf import settings from django.conf import settings
from utils.logger_filter import decorator_request_logger from utils.logger_filter import decorator_request_logger
from utils.tracecalls import decorator_logger
APP_ENV = os.getenv('APP_ENV') APP_ENV = os.getenv('APP_ENV')
if APP_ENV == 'dev': if APP_ENV == 'dev':
@ -19,7 +20,7 @@ else:
DING_CORP_ID, TITLE DING_CORP_ID, TITLE
msg_template = 'messages.html' msg_template = 'messages.html'
logger = logging.getLogger('django') logger = logging.getLogger(__name__)
class PARAMS(object): class PARAMS(object):
@ -43,7 +44,7 @@ scan_params = PARAMS()
_ops = scan_params.ops _ops = scan_params.ops
@decorator_request_logger(logger) @decorator_logger(logger, log_head='Request', pretty=True, indent=2, verbose=1)
def auth(request): def auth(request):
home_url = '%s://%s' % (request.scheme, HOME_URL) home_url = '%s://%s' % (request.scheme, HOME_URL)
corp_id = scan_params.corp_id corp_id = scan_params.corp_id
@ -56,25 +57,13 @@ def auth(request):
global_title = TITLE global_title = TITLE
if request.method == 'GET': if request.method == 'GET':
code = request.GET.get('code', None)
username = request.GET.get('username', None)
# 如果满足，说明已经授权免密登录
if username and code and request.session.get(username) == code:
context = {'global_title': TITLE,
'username': username,
'code': code,
}
return render(request, 'reset_password.html', context)
return render(request, 'auth.html', locals()) return render(request, 'auth.html', locals())
else: else:
logger.error('[异常] 请求方法：%s，请求路径%s' % (request.method, request.path)) logger.error('[异常] 请求方法：%s，请求路径%s' % (request.method, request.path))
@decorator_request_logger(logger) @decorator_logger(logger, log_head='Request', pretty=True, indent=2, verbose=1)
def index(request): def index(request):
"""
用户自行修改密码/首页
"""
home_url = '%s://%s' % (request.scheme, HOME_URL) home_url = '%s://%s' % (request.scheme, HOME_URL)
scan_app = scan_params.AUTH_APP scan_app = scan_params.AUTH_APP
global_title = TITLE global_title = TITLE
@ -93,41 +82,45 @@ def index(request):
else: else:
_msg = check_form _msg = check_form
logger.error('[异常] 请求方法：%s，请求路径：%s，错误信息：%s' % (request.method, request.path, _msg)) logger.error('[异常] 请求方法：%s，请求路径：%s，错误信息：%s' % (request.method, request.path, _msg))
context = {'global_title': TITLE, context = {
'msg': _msg, 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': _msg,
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
# 格式化用户名 # 格式化用户名
_, username = format2username(username) _, username = format2username(username)
if _ is False: if _ is False:
context = {'global_title': TITLE, context = {
'msg': username, 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': username,
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
# 检测账号状态 # 检测账号状态
auth_status, auth_result = AdOps().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 = {'global_title': TITLE, context = {
'msg': str(auth_result), 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': str(auth_result),
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
return ops_account(AdOps(), request, msg_template, home_url, username, new_password) return ops_account(AdOps(), request, msg_template, home_url, username, new_password)
else: else:
context = {'global_title': TITLE, context = {
'msg': "不被接受的认证信息，请重新尝试认证授权。", 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': "不被接受的认证信息，请重新尝试认证授权。",
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
@decorator_request_logger(logger) @decorator_logger(logger, log_head='Request', pretty=True, indent=2, verbose=1)
def reset_password(request): def reset_password(request):
""" """
钉钉扫码并验证信息通过之后，在重置密码页面将用户账号进行绑定 钉钉扫码并验证信息通过之后，在重置密码页面将用户账号进行绑定
@ -140,21 +133,23 @@ def reset_password(request):
username = request.GET.get('username', None) username = request.GET.get('username', None)
# 如果满足，说明已经授权免密登录 # 如果满足，说明已经授权免密登录
if username and code and request.session.get(username) == code: if username and code and request.session.get(username) == code:
context = {'global_title': TITLE, context = {
'username': username, 'global_title': TITLE,
'code': code, 'username': username,
} 'code': code,
}
return render(request, 'reset_password.html', context) return render(request, 'reset_password.html', context)
else: else:
if code: if code:
logger.info('[成功] 请求方法：%s，请求路径：%s，Code：%s' % (request.method, request.path, code)) logger.info('[成功] 请求方法：%s，请求路径：%s，Code：%s' % (request.method, request.path, code))
else: else:
logger.error('[异常] 请求方法：%s，请求路径：%s，未能拿到Code。' % (request.method, request.path)) logger.error('[异常] 请求方法：%s，请求路径：%s，未能拿到Code。' % (request.method, request.path))
context = {'global_title': TITLE, context = {
'msg': "错误，临时授权码己失效，请尝试重新认证授权..", 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': "错误，临时授权码己失效，请尝试重新认证授权..",
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
try: try:
# 用code换取用户基本信息 # 用code换取用户基本信息
@ -164,86 +159,91 @@ def reset_password(request):
# 账号在企业微信或钉钉中是否是激活的 # 账号在企业微信或钉钉中是否是激活的
_, res = get_user_is_active(user_info) _, res = get_user_is_active(user_info)
if not _: if not _:
context = {'global_title': TITLE, context = {
'msg': '当前扫码的用户未激活或可能己离职，用户信息如下：%s' % user_info, 'global_title': TITLE,
'button_click': "window.location.href='%s'" % home_url, 'msg': '当前扫码的用户未激活或可能己离职，用户信息如下：%s' % user_info,
'button_display': "返回主页" 'button_click': "window.location.href='%s'" % home_url,
} 'button_display': "返回主页"
}
return render(request, msg_template, context) return render(request, msg_template, context)
except Exception as callback_e: except Exception as callback_e:
context = {'global_title': TITLE, context = {
'msg': "错误[%s]，请与管理员联系." % str(callback_e), 'global_title': TITLE,
'button_click': "window.location.href='%s'" % home_url, 'msg': "错误[%s]，请与管理员联系." % str(callback_e),
'button_display': "返回主页" 'button_click': "window.location.href='%s'" % home_url,
} 'button_display': "返回主页"
}
logger.error('[异常] ：%s' % str(callback_e)) logger.error('[异常] ：%s' % str(callback_e))
return render(request, msg_template, context) return render(request, msg_template, context)
# 通过user_info拿到用户邮箱，并格式化为username # 通过user_info拿到用户邮箱，并格式化为username
_, email = get_email_from_userinfo(user_info) _, email = get_email_from_userinfo(user_info)
if not _: if not _:
context = {'global_title': TITLE, context = {
'msg': email, 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': email,
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
_, username = format2username(email) _, username = format2username(email)
if _ is False: if _ is False:
context = {'global_title': TITLE, context = {
'msg': username, 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': username,
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
# 如果邮箱能提取到，则格式化之后，提取出账号提交到前端绑定
if username: if username:
request.session[username] = code request.session[username] = code
context = {'global_title': TITLE, context = {
'username': username, 'global_title': TITLE,
'code': code, 'username': username,
} 'code': code,
}
return render(request, 'reset_password.html', context) return render(request, 'reset_password.html', context)
else: else:
context = {'global_title': TITLE, context = {
'msg': "{}，您好，企业{}中未能找到您账号的邮箱配置，请联系HR完善信息。".format( 'global_title': TITLE,
user_info.get('name'), scan_params.AUTH_APP), 'msg': "{}，您好，企业{}中未能找到您账号的邮箱配置，请联系HR完善信息。".format(
'button_click': "window.location.href='%s'" % '/auth', user_info.get('name'), scan_params.AUTH_APP),
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
# 重置密码页面，输入新密码后点击提交 # 重置密码页面，输入新密码后点击提交
elif request.method == 'POST': elif request.method == 'POST':
username = request.POST.get('username') username = request.POST.get('username')
code = request.POST.get('code') code = request.POST.get('code')
if code and request.session.get(username) == code: if username and code and request.session.get(username) == code:
_new_password = request.POST.get('new_password').strip() _new_password = request.POST.get('new_password').strip()
try: try:
return ops_account(ad_ops=AdOps(), request=request, msg_template=msg_template, home_url=home_url, return ops_account(ad_ops=AdOps(), request=request, msg_template=msg_template, home_url=home_url,
username=username, new_password=_new_password) username=username, new_password=_new_password)
except Exception as reset_e: except Exception as reset_e:
context = {'global_title': TITLE, context = {
'msg': "错误[%s]，请与管理员联系." % str(reset_e), 'global_title': TITLE,
'button_click': "window.location.href='%s'" % home_url, 'msg': "错误[%s]，请与管理员联系." % str(reset_e),
'button_display': "返回主页" 'button_click': "window.location.href='%s'" % home_url,
} 'button_display': "返回主页"
}
logger.error('[异常] ：%s' % str(reset_e)) logger.error('[异常] ：%s' % str(reset_e))
return render(request, msg_template, context) return render(request, msg_template, context)
finally: else:
del request.session[username] context = {
else: 'global_title': TITLE,
context = {'global_title': TITLE, 'msg': "认证已经失效，可尝试从重新认证授权。",
'msg': "认证已经失效，可尝试从重新认证授权。", 'button_click': "window.location.href='%s'" % '/auth',
'button_click': "window.location.href='%s'" % '/auth', 'button_display': "重新认证授权"
'button_display': "重新认证授权" }
} return render(request, msg_template, context)
return render(request, msg_template, context)
@decorator_request_logger(logger) @decorator_logger(logger, log_head='Request', pretty=True, indent=2, verbose=1)
def unlock_account(request): def unlock_account(request):
""" """
解锁账号 解锁账号
@ -255,53 +255,56 @@ def unlock_account(request):
if request.method == 'GET': if request.method == 'GET':
code = request.GET.get('code') code = request.GET.get('code')
username = request.GET.get('username') username = request.GET.get('username')
if code and request.session.get(username) == code: if username and code and request.session.get(username) == code:
context = {'global_title': TITLE, context = {
'username': username, 'global_title': TITLE,
'code': code, 'username': username,
} 'code': code,
}
return render(request, 'unlock.html', context) return render(request, 'unlock.html', context)
else: else:
context = {'global_title': TITLE, context = {
'msg': "{}，您好，当前会话可能已经过期，请再试一次吧。".format(username), 'global_title': TITLE,
'button_click': "window.location.href='%s'" % '/auth', 'msg': "{}，您好，当前会话可能已经过期，请再试一次吧。".format(username),
'button_display': "重新认证授权" 'button_click': "window.location.href='%s'" % '/auth',
} 'button_display': "重新认证授权"
}
return render(request, msg_template, context) return render(request, msg_template, context)
if request.method == 'POST': if request.method == 'POST':
username = request.POST.get('username') username = request.POST.get('username')
code = request.POST.get('code') code = request.POST.get('code')
if request.session.get(username) and request.session.get(username) == code: if username and code and request.session.get(username) == code:
try: try:
return ops_account(AdOps(), request, msg_template, home_url, username, None) return ops_account(AdOps(), request, msg_template, home_url, username, None)
except Exception as reset_e: except Exception as reset_e:
context = {'global_title': TITLE, context = {
'msg': "错误[%s]，请与管理员联系." % str(reset_e), 'global_title': TITLE,
'button_click': "window.location.href='%s'" % home_url, 'msg': "错误[%s]，请与管理员联系." % str(reset_e),
'button_display': "返回主页" 'button_click': "window.location.href='%s'" % home_url,
} 'button_display': "返回主页"
}
logger.error('[异常] ：%s' % str(reset_e)) logger.error('[异常] ：%s' % str(reset_e))
return render(request, msg_template, context) return render(request, msg_template, context)
finally: else:
del request.session[username] context = {
else: 'global_title': TITLE,
context = {'global_title': TITLE, 'msg': "认证已经失效，请尝试从重新进行认证授权。",
'msg': "认证已经失效，请尝试从重新进行认证授权。", 'button_click': "window.location.href='%s'" % '/auth',
'button_click': "window.location.href='%s'" % '/auth', 'button_display': "重新认证授权"
'button_display': "重新认证授权" }
} return render(request, msg_template, context)
return render(request, msg_template, context)
@decorator_request_logger(logger) @decorator_logger(logger, log_head='Request', pretty=True, indent=2, verbose=1)
def messages(request): def messages(request):
_msg = request.GET.get('msg') _msg = request.GET.get('msg')
button_click = request.GET.get('button_click') button_click = request.GET.get('button_click')
button_display = request.GET.get('button_display') button_display = request.GET.get('button_display')
context = {'global_title': TITLE, context = {
'msg': _msg, 'global_title': TITLE,
'button_click': button_click, 'msg': _msg,
'button_display': button_display 'button_click': button_click,
} 'button_display': button_display
}
return render(request, msg_template, context) return render(request, msg_template, context)

2
templates/unlock.html
View File

@ -3,7 +3,7 @@
{% block paneltitle %}解锁账号{% endblock %} {% block paneltitle %}解锁账号{% endblock %}
{% block middleblock %} {% block middleblock %}
<div class="layui-row"> <div class="layui-row">
<form class="layui-form layui-form-pane" name="unlockAccount" method="post" action="unlockAccount" autocomplete="off">{% csrf_token %} <form class="layui-form layui-form-pane" name="unlockAccount" method="post" action="" autocomplete="off">{% csrf_token %}
<div class="layui-form-item"> <div class="layui-form-item">
<label class="layui-form-label">账号</label> <label class="layui-form-label">账号</label>
<div class="layui-input-block"> <div class="layui-input-block">

22
utils/ad_ops.py
View File

@ -4,6 +4,8 @@ from ldap3.core.exceptions import LDAPInvalidCredentialsResult, LDAPOperationRes
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
from utils.tracecalls import decorator_logger
import logging
APP_ENV = os.getenv('APP_ENV') APP_ENV = os.getenv('APP_ENV')
if APP_ENV == 'dev': if APP_ENV == 'dev':
@ -11,6 +13,8 @@ if APP_ENV == 'dev':
else: else:
from conf.local_settings import * from conf.local_settings import *
logger = logging.getLogger(__name__)
""" """
根据以下网站的说明： 根据以下网站的说明：
https://docs.microsoft.com/zh-cn/troubleshoot/windows/win32/change-windows-active-directory-user-password https://docs.microsoft.com/zh-cn/troubleshoot/windows/win32/change-windows-active-directory-user-password
@ -86,6 +90,7 @@ class AdOps(object):
except LDAPException as l_e: except LDAPException as l_e:
return False, LDAPException("LDAPException: " + str(l_e)) return False, LDAPException("LDAPException: " + str(l_e))
@decorator_logger(logger, log_head='AdOps', pretty=True, indent=2, verbose=1)
def ad_auth_user(self, username, password): def ad_auth_user(self, username, password):
""" """
验证账号 验证账号
@ -141,6 +146,7 @@ class AdOps(object):
except Exception as e: except Exception as e:
return False, "AdOps Exception: {}".format(e) return False, "AdOps Exception: {}".format(e)
@decorator_logger(logger, log_head='AdOps', pretty=True, indent=2, verbose=1)
def ad_get_user_displayname_by_account(self, username): def ad_get_user_displayname_by_account(self, username):
""" """
通过username查询某个用户的显示名 通过username查询某个用户的显示名
@ -156,6 +162,7 @@ class AdOps(object):
except Exception as e: except Exception as e:
return False, "AdOps Exception: {}".format(e) return False, "AdOps Exception: {}".format(e)
@decorator_logger(logger, log_head='AdOps', pretty=True, indent=2, verbose=1)
def ad_get_user_dn_by_account(self, username): def ad_get_user_dn_by_account(self, username):
""" """
通过username查询某个用户的完整DN 通过username查询某个用户的完整DN
@ -168,10 +175,14 @@ class AdOps(object):
attributes=['distinguishedName']) attributes=['distinguishedName'])
return True, str(self.conn.entries[0]['distinguishedName']) return True, str(self.conn.entries[0]['distinguishedName'])
except IndexError: except IndexError:
logger.error("AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。")
logger.error("self.conn.search(BASE_DN, {}, attributes=['distinguishedName'])".format(SEARCH_FILTER.format(username)))
return False, "AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。" return False, "AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。"
except Exception as e: except Exception as e:
logger.error("AdOps Exception: {}".format(e))
return False, "AdOps Exception: {}".format(e) return False, "AdOps Exception: {}".format(e)
@decorator_logger(logger, log_head='AdOps', pretty=True, indent=2, verbose=1)
def ad_get_user_status_by_account(self, username): def ad_get_user_status_by_account(self, username):
""" """
通过username查询某个用户的账号状态 通过username查询某个用户的账号状态
@ -180,14 +191,18 @@ class AdOps(object):
""" """
try: try:
self.__conn() self.__conn()
self.conn.search(BASE_DN, SEARCH_FILTER.format(username), self.conn.search(BASE_DN, SEARCH_FILTER.format(username), attributes=['userAccountControl'])
attributes=['userAccountControl'])
return True, self.conn.entries[0]['userAccountControl'] return True, self.conn.entries[0]['userAccountControl']
except IndexError: except IndexError:
logger.error("AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。")
logger.error("self.conn.search({}, {}, attributes=['userAccountControl'])".format(BASE_DN, SEARCH_FILTER.format(username)))
logger.info("self.conn.entries -- {}".format(self.conn.entries))
return False, "AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。" return False, "AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。"
except Exception as e: except Exception as e:
logger.error("AdOps Exception: {}".format(e))
return False, "AdOps Exception: {}".format(e) return False, "AdOps Exception: {}".format(e)
@decorator_logger(logger, log_head='AdOps', pretty=True, indent=2, verbose=1)
def ad_unlock_user_by_account(self, username): def ad_unlock_user_by_account(self, username):
""" """
通过username解锁某个用户 通过username解锁某个用户
@ -201,10 +216,12 @@ class AdOps(object):
except IndexError: except IndexError:
return False, "AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。" return False, "AdOps Exception: Connect.search未能检索到任何信息，当前账号可能被排除在<SEARCH_FILTER>之外，请联系管理员处理。"
except Exception as e: except Exception as e:
logger.error("AdOps Exception: {}".format(e))
return False, "AdOps Exception: {}".format(e) return False, "AdOps Exception: {}".format(e)
else: else:
return False, user_dn return False, user_dn
@decorator_logger(logger, log_head='AdOps', pretty=True, indent=2, verbose=1)
def ad_reset_user_pwd_by_account(self, username, new_password): def ad_reset_user_pwd_by_account(self, username, new_password):
""" """
重置某个用户的密码 重置某个用户的密码
@ -242,6 +259,7 @@ class AdOps(object):
else: else:
return False, user_dn return False, user_dn
@decorator_logger(logger, log_head='AdOps', pretty=True, indent=2, verbose=1)
def ad_get_user_locked_status_by_account(self, username): def ad_get_user_locked_status_by_account(self, username):
""" """
通过username获取某个用户账号是否被锁定 通过username获取某个用户账号是否被锁定

149
utils/tracecalls.py Normal file
View File

@ -0,0 +1,149 @@
# -*- coding: utf-8 -*-
import os.path
import re
import sys
from copy import deepcopy as dcopy
from functools import wraps
from traceback import format_exc
from pwdselfservice.settings import BASE_DIR
from pprint import pformat as cformat
NOT_CHECK_CALL_FUNC_NAME = []
DEBUG_FLAG_FILE_PATH = os.path.join(BASE_DIR, '/log')
class TraceFuncContext:
def __init__(self, func_name, logger, log_head='Run function', debug_flag_name=None, verbose=1,
pretty=False, indent=0, check_calls=None):
self.name = func_name
self.logger = logger
self.debug_flag_name = debug_flag_name
self.log_head = log_head
self.verbose = verbose
self.pretty = pretty
self.indent = indent
self.check_calls = check_calls
self.copied_verbose = dcopy(verbose)
self.is_exit = False
self.debug_flag_suffix = '.debug.flag'
if self.verbose is not None:
self.check_calls = None
def __enter__(self):
sys.settrace(self.get_callbacks)
def check_debug_flag(self):
if self.debug_flag_name:
if os.path.isfile(os.path.join(DEBUG_FLAG_FILE_PATH, '{}{}'.format(
self.debug_flag_name, self.debug_flag_suffix))):
self.verbose = 2
else:
self.verbose = self.copied_verbose
@staticmethod
def check_in_excludes(func_name):
if len(NOT_CHECK_CALL_FUNC_NAME) > 0:
regex_gen = r'{0}'.format('|'.join(NOT_CHECK_CALL_FUNC_NAME))
re_c = re.compile(regex_gen, flags=re.IGNORECASE)
return re_c.search(func_name)
return False
def get_callbacks(self, frame, event, arg=None):
self.check_debug_flag()
__co_name = frame.f_code.co_name
if event != 'call': # Only trace call
return
if self.verbose:
if self.verbose == 1 or self.verbose == 'v':
self.check_calls = [self.name]
if __co_name not in self.check_calls:
return
elif self.verbose == 2 or self.verbose == 'vv':
if self.check_calls is None:
self.check_calls = list(set(frame.f_code.co_names))
self.check_calls.append(self.name)
if __co_name not in self.check_calls or self.check_in_excludes(__co_name):
return
else:
raise ValueError("UNKNOWN VERBOSE VALUE: support verbose is 1/2 or v/vv...")
else:
if self.check_calls is None:
self.check_calls = [self.name]
else:
if isinstance(self.check_calls, list):
self.check_calls.append(self.name)
else:
raise ValueError("CHECK CALLS TYPE ERROR: check_calls need a list or tuple with function name.")
if __co_name not in self.check_calls:
return
return self.get_code_line
def get_code_line(self, frame, event, agr=None):
# 正常情况下只有line或return事件才做记录
if event not in ['line', 'return']:
return
__code = frame.f_code
__func_name = __code.co_name
__line_num = frame.f_lineno
__locals = frame.f_locals
if __func_name == self.name:
self.logger.info(
"{4} [{0}] trace detail "
"--- {1} {2}, locals as following: {5}{3}{5}".format(__func_name,
event,
__line_num,
cformat(__locals, indent=self.indent)
if self.pretty else __locals,
self.log_head,
'\n' if self.pretty else ''
))
else:
self.logger.info(
"{5} [{0}] call [{1}] trace detail "
"--- {2} {3}, locals as following: {6}{4}{6}".format(self.name,
__func_name,
event,
__line_num,
cformat(__locals, indent=self.indent)
if self.pretty else __locals,
self.log_head,
'\n' if self.pretty else ''
))
def __exit__(self, exc_type, exc_val, exc_tb):
sys.settrace(None)
def decorator_logger(logger, log_head='Run function', debug_flag_name=None, verbose=1, check_calls=None,
pretty=False, indent=0):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
func_consts = func.__code__.co_consts
logger.debug("{4} [{0}] entering trace with --- consts-{3}, args-{1}, kwargs-{2}...".format(
func.__name__, args, kwargs, func_consts, log_head))
with TraceFuncContext(func.__name__, logger,
log_head=log_head,
debug_flag_name=debug_flag_name,
verbose=verbose,
check_calls=check_calls,
pretty=pretty,
indent=indent
):
try:
func_res = func(*args, **kwargs)
logger.debug("{4} [{0}] exiting trace with --- consts-{3}, args-{1}, kwargs-{2}...".format(
func.__name__, args, kwargs, func_consts, log_head))
return func_res
except Exception as e:
logger.error(
"{4} [{0}] has exception, trace with --- consts-{3}, args-{1}, kwargs-{2}. Trackback as "
"following: ".format(
func.__name__, args, kwargs, func_consts, log_head))
logger.error(format_exc())
raise e
return wrapper
return decorator