多因素身份认证 (MFA) 插件：手机验证码认证因素配置流程
作者：龙归科技
- 2022 年 9 月 14 日 江苏
本文字数：15747 字
阅读完需：约 52 分钟
功能说明
对用户表扩展手机号码字段，允许用户通过手机号码与验证码的方式进行认证，注册，重置密码以及更换手机号。多因素身份认证 (MFA) 是保护企业 IT 资源访问安全的一种关键工具，也是零信任安全模型的核心组成。特别在远程办公以及数据泄露事件层出不穷的背景下，越来越多企业都开始考虑实施多因素身份认证策略。根据近期一项对 IT 专业人员的调查，52.6%的中小型企业已经要求为所有应用程序和系统登录添加多因素身份认证。其中手机推送认证形式（无密码认证）的二次认证是保证良好用户体验的绝佳选择。开源 IDaaS 方舟一账通 ArkID 系统内置手机验证码认证插件简单配置、马上可用，降低实施成本，提升应用集成效率。
普通用户
• 在 “注册” 页面实现手机号码+验证码用户注册
• 在 “登录” 页面实现手机号码+验证码用户登录
• 在 “更改密码” 页面实现手机号码+验证码方式下密码更改
• 在 “我的 - 认证管理“ 中添加重置手机号码的功能
租户管理员
• 在”用户管理 - 用户列表“中编辑页面添加手机号码编辑功能前置条件需配合短信插件使用，系统已默认提供阿里云短信插件，如需查看配置方法请移步阿里云短信插件文档。
配置指南
01 插件租赁
经由左侧菜单栏依次进入【租户管理】->【插件管理】，在插件租赁页面中找到手机验证码认证因素插件卡片，点击租赁
02 认证因素配置
经由左侧菜单栏依次进入【认证管理】-> 【认证因素】,点击创建按钮，类型选择"mobile",短信插件运行时选择一个合适的已配置的短信插件运行时配置，至此配置完成
03 登录界面
04 注册界面
05 密码修改界面
06 更换手机号码界面
由用户头像菜单进入【认证管理】界面,选择更改手机号码标签页
实现思路
普通用户：手机号码+验证码用户注册/登录/重置密码：
普通用户：重置手机号码：
管理员用户： 更换用户手机号码
抽象方法实现
load
authenticate
register
reset_password
create_login_page
create_register_page
create_password_page
create_other_page
create_auth_manage_page
check_auth_data
fix_login_page
代码
extension_root.com_longgui_auth_factor_mobile.MobileAuthFactorExtension (AuthFactorExtension)
手机短信验证码认证因素插件
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
self.create_extension_config_schema()
self.register_extend_field(UserMobile, "mobile") from api.v1.schema.auth import AuthIn from api.v1.schema.user import UserCreateIn,UserItemOut,UserUpdateIn,UserListItemOut from api.v1.schema.mine import ProfileSchemaOut self.register_extend_api( AuthIn, UserCreateIn, UserItemOut, UserUpdateIn, UserListItemOut, mobile=(Optional[str],Field(title=_("电话号码"))), # areacode=(str,Field(title=_("区号"))) ) self.register_extend_api( ProfileSchemaOut, mobile=(Optional[str],Field(readnotallow=True)) )
# 注册发送短信接口 self.url_send_sms_code = self.register_api( '/config/{config_id}/send_sms_code/', 'POST', self.send_sms_code, tenant_path=True, auth=None, respnotallow=SendSMSCodeOut, ) print(self.url_send_sms_code)
def authenticate(self, event, **kwargs): """ 认证
通过手机号码查找用户并校验短信验证码
Args: event (Event): 事件
""" tenant = event.tenant request = event.request data = request.POST or json.load(request.body)
mobile = data.get('mobile') sms_code = data.get('sms_code')
user = User.expand_objects.filter(tenant=tenant,mobile=mobile) if len(user) > 1: logger.error(f'{mobile}在数据库中匹配到多个用户') return self.auth_failed(event, data=self.error(ErrorCode.CONTACT_MANAGER)) if user: user = user[0] if check_sms_code(mobile, sms_code): user = User.active_objects.get(id=user.get("id")) return self.auth_success(user,event) else: msg = ErrorCode.SMS_CODE_MISMATCH else: msg = ErrorCode.MOBILE_NOT_EXISTS_ERROR return self.auth_failed(event, data=self.error(msg))
@transaction.atomic()def register(self, event, **kwargs): """ 注册用户
Args: event (Event): 事件 """ tenant = event.tenant request = event.request data = request.POST or json.load(request.body)
mobile = data.get('mobile') sms_code = data.get('sms_code') username = data.get('username')
config = self.get_current_config(event) ret, message = self.check_mobile_exists(mobile, tenant) if not ret: return self.error(message)
if not check_sms_code(mobile, sms_code): return self.error(ErrorCode.SMS_CODE_MISMATCH)
ret, message = self.check_username_exists(username, tenant) if not ret: return self.error(message)
user = User(tenant=tenant)
user.mobile = mobile user.username = username
user.save() tenant.users.add(user) tenant.save()
return user
def reset_password(self, event, **kwargs): """ 重置密码
Args: event (Event): 事件 """ tenant = event.tenant request = event.request data = request.POST or json.load(request.body)
mobile = data.get('mobile') sms_code = data.get('sms_code')
password = data.get('password') checkpassword = data.get('checkpassword')
if password != checkpassword: return self.error(ErrorCode.PASSWORD_IS_INCONSISTENT)
if not check_sms_code(mobile, sms_code): return self.error(ErrorCode.SMS_CODE_MISMATCH)
user = User.expand_objects.filter(tenant=tenant,mobile=mobile)
if len(user) > 1: logger.error(f'{mobile}在数据库中匹配到多个用户') return self.error(ErrorCode.CONTACT_MANAGER) if user: user = user[0] user = User.active_objects.get(id=user.get("id")) user.password = make_password(password) user.save() return self.success()
return self.error(ErrorCode.MOBILE_NOT_EXISTS_ERROR)
def create_login_page(self, event, config, config_data): """ 生成手机验证码登录页面Schema描述
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """
items = [ { "type": "text", "name":"mobile", "placeholder": "手机号码", "append": { "title": "发送验证码", "http": { "url": self.url_send_sms_code, "method": "post", "params": { "mobile": "mobile", "areacode": "86", }, }, "delay": 60 } }, { "type": "text", "name":"sms_code", "placeholder": "验证码", } ] self.add_page_form(config, self.LOGIN, "手机验证码登录", items, config_data)
def create_register_page(self, event, config, config_data): """生成手机验证码用户注册页面Schema描述
因本插件提供重置密码功能，此处需用户指定账号用户名
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """ items = [ { "type": "text", "name": "username", "placeholder": "用户名" }, { "type": "text", "name":"mobile", "placeholder": "手机号码", "append": { "title": "发送验证码", "http": { "url": self.url_send_sms_code, "method": "post", "params": { "mobile": "mobile", "areacode": "86", }, }, "delay": 60 } }, { "type": "text", "name":"sms_code", "placeholder": "验证码" } ] self.add_page_form(config, self.REGISTER, "手机验证码注册", items, config_data)
def create_password_page(self, event, config, config_data): """生成重置密码页面Schema描述
通过手机验证码重置密码时需提供手机号码以及对应验证码，同时此处添加新密码确认机制
注意：重置密码功能需要启用用户名密码认证插件以提供完整支持
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """ items = [ { "type": "text", "name":"mobile", "placeholder": "手机号码", "append": { "title": "发送验证码", "http": { "url": self.url_send_sms_code, "method": "post", "params": { "mobile": "mobile", "areacode": "86", }, }, } }, { "type": "text", "name":"sms_code", "placeholder": "验证码" }, { "type": "password", "name":"password", "placeholder": "密码" }, { "type": "password", "name":"checkpassword", "placeholder": "密码确认" } ] self.add_page_form(config, self.RESET_PASSWORD, "手机验证码重置密码", items, config_data)
def create_other_page(self, event, config, config_data): """创建其他页面（本插件无相关页面）
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """ pass
def check_mobile_exists(self, mobile, tenant): """检查电话号码是否已存在
Args: mobile (str): 手机号 tenant (Tenant): 租户
Returns: (bool,ErrorCode): mobile是否存在以及对应错误 """ if not mobile: return False, ErrorCode.MOBILE_EMPTY
if User.expand_objects.filter(tenant=tenant,mobile=mobile).count(): return False, ErrorCode.MOBILE_EXISTS_ERROR return True, None
def check_username_exists(self,username,tenant): """检查用户名是否已存在
Args: username (str): 用户名 tenant (Tenant): 租户
Returns: (bool,ErrorCode): username是否存在以及对应错误 """ # 检查username是否为空 if not username: return False, ErrorCode.USERNAME_EMPTY # 检查username是否已存在 if User.expand_objects.filter(tenant=tenant,username=username).count(): return False, ErrorCode.USERNAME_EXISTS_ERROR
return True, None
def check_auth_data(self, event, **kwargs): pass
def fix_login_page(self, event, **kwargs): pass
def create_auth_manage_page(self): """ 创建“我的-认证管理”中的更换手机号码页面 """ _pages = []
mine_mobile_path = self.register_api( "/mine_mobile/", "GET", self.mine_mobile, tenant_path=True, auth=GlobalAuth(), respnotallow=MineMobileOut )
upodate_mine_mobile_path = self.register_api( "/mine_mobile/", 'POST', self.update_mine_mobile, tenant_path=True, auth=GlobalAuth(), respnotallow=UpdateMineMobileOut )
name = '更改手机号码'
page = pages.FormPage(name=name) page.create_actions( init_actinotallow=actions.DirectAction( path=mine_mobile_path, method=actions.FrontActionMethod.GET, ), global_actinotallow={ 'confirm': actions.ConfirmAction( path=upodate_mine_mobile_path ), } )
_pages.append(page) return _pages
def create_extension_config_schema(self): """创建插件运行时配置schema描述 """ select_sms_page = pages.TablePage(select=True,name=_("指定短信插件运行时"))
self.register_front_pages(select_sms_page)
select_sms_page.create_actions( init_actinotallow=actions.DirectAction( path='/api/v1/tenants/{tenant_id}/config_select/?extension__type=sms', method=actions.FrontActionMethod.GET ) )
MobileAuthFactorSchema = create_extension_schema( 'MobileAuthFactorSchema', __file__, [ ( 'sms_config', MobileAuthFactorConfigSchema, Field( title=_('sms extension config', '短信插件运行时'), page=select_sms_page.tag, ), ), ( 'code_length', int, Field( title=_('code_length', '验证码长度'), default=6 ) ), ( 'expired', Optional[int], Field( title=_('expired', '有效期/分钟'), default=10, ) ), ], BaseAuthFactorSchema, ) self.register_auth_factor_schema(MobileAuthFactorSchema, 'mobile')
@operation(SendSMSCodeOut)def send_sms_code(self,request,tenant_id,config_id:str,data:SendSMSCodeIn): """发送短信验证码 """ tenant = request.tenant mobile = data.mobile config = self.get_config_by_id(config_id) if not config: return self.error(ErrorCode.CONFIG_IS_NOT_EXISTS)
if not mobile or mobile=="mobile": return self.error(ErrorCode.MOBILE_EMPTY)
code = create_sms_code(tenant,mobile,config.config.get('code_length',6),config.config.get("expired",10)*60)
responses = dispatch_event( Event( tag=SEND_SMS, tenant=tenant, request=request, data={ "config_id":config.config["sms_config"]["id"], "mobile":data.mobile, "code": code, "areacode": data.areacode, "username": request.user.username if request.user else "" }, packages=config.config["sms_config"]["package"]
) )
if not responses: return self.error(ErrorCode.SMS_EXTENSION_NOT_EXISTS) useless, (data, extension) = responses[0] if data: return self.success() else: return self.error(ErrorCode.SMS_SEND_FAILED)
@operation(UpdateMineMobileOut,roles=[TENANT_ADMIN, PLATFORM_ADMIN, NORMAL_USER])def update_mine_mobile(self, request, tenant_id: str,data:UpdateMineMobileIn): """ 普通用户：更新手机号码 """ mobile = data.mobile ret, message = self.check_mobile_exists(mobile, request.tenant) if not ret: return self.error(message)
if not check_sms_code(request.tenant,mobile,data.code): return self.error(ErrorCode.SMS_CODE_MISMATCH)
user = request.user user.mobile=data.mobile user.save()
return self.success()
@operation(MineMobileOut,roles=[TENANT_ADMIN, PLATFORM_ADMIN, NORMAL_USER])def mine_mobile(self,request,tenant_id: str): user = request.user user_expand = User.expand_objects.filter(id=user.id).first()
config = self.get_tenant_configs(request.tenant).first()
if not config: return self.error( ErrorCode.CONFIG_IS_NOT_EXISTS )
return self.success( data={ "current_mobile": user_expand.get("mobile",None), "mobile": "", "code": "", "config_id": config.id.hex, }, )authenticate(self, event, **kwargs)
复制代码
认证通过手机号码查找用户并校验短信验证码 Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/init.pydef authenticate(self, event, **kwargs):""" 认证
通过手机号码查找用户并校验短信验证码
Args: event (Event): 事件
"""tenant = event.tenantrequest = event.requestdata = request.POST or json.load(request.body)
mobile = data.get('mobile')sms_code = data.get('sms_code')
user = User.expand_objects.filter(tenant=tenant,mobile=mobile)if len(user) > 1: logger.error(f'{mobile}在数据库中匹配到多个用户') return self.auth_failed(event, data=self.error(ErrorCode.CONTACT_MANAGER))if user: user = user[0] if check_sms_code(mobile, sms_code): user = User.active_objects.get(id=user.get("id")) return self.auth_success(user,event) else: msg = ErrorCode.SMS_CODE_MISMATCHelse: msg = ErrorCode.MOBILE_NOT_EXISTS_ERRORreturn self.auth_failed(event, data=self.error(msg))
复制代码
authenticate(self, event, **kwargs)
认证
通过手机号码查找用户并校验短信验证码
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/__init__.py
def authenticate(self, event, **kwargs): """ 认证
通过手机号码查找用户并校验短信验证码
Args: event (Event): 事件
""" tenant = event.tenant request = event.request data = request.POST or json.load(request.body)
mobile = data.get('mobile') sms_code = data.get('sms_code')
user = User.expand_objects.filter(tenant=tenant,mobile=mobile) if len(user) > 1: logger.error(f'{mobile}在数据库中匹配到多个用户') return self.auth_failed(event, data=self.error(ErrorCode.CONTACT_MANAGER)) if user: user = user[0] if check_sms_code(mobile, sms_code): user = User.active_objects.get(id=user.get("id")) return self.auth_success(user,event) else: msg = ErrorCode.SMS_CODE_MISMATCH else: msg = ErrorCode.MOBILE_NOT_EXISTS_ERROR return self.auth_failed(event, data=self.error(msg))
复制代码
check_auth_data(self, event, **kwargs)
响应检查认证凭证事件
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/__init__.py
def check_auth_data(self, event, **kwargs): pass复制代码
check_mobile_exists(self, mobile, tenant) #
检查电话号码是否已存在
Parameters:
Returns:
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def check_mobile_exists(self, mobile, tenant): """检查电话号码是否已存在
Args: mobile (str): 手机号 tenant (Tenant): 租户
Returns: (bool,ErrorCode): mobile是否存在以及对应错误 """ if not mobile: return False, ErrorCode.MOBILE_EMPTY
if User.expand_objects.filter(tenant=tenant,mobile=mobile).count(): return False, ErrorCode.MOBILE_EXISTS_ERROR return True, None复制代码
check_username_exists(self, username, tenant)
检查用户名是否已存在
Parameters:
Returns:
Source code in extension_root/com_longgui_auth_factor_mobile/__init__.py
def check_username_exists(self,username,tenant): """检查用户名是否已存在
Args: username (str): 用户名 tenant (Tenant): 租户
Returns: (bool,ErrorCode): username是否存在以及对应错误 """ # 检查username是否为空 if not username: return False, ErrorCode.USERNAME_EMPTY # 检查username是否已存在 if User.expand_objects.filter(tenant=tenant,username=username).count(): return False, ErrorCode.USERNAME_EXISTS_ERROR
return True, None
复制代码
create_auth_manage_page(self)
创建“我的-认证管理”中的更换手机号码页面
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def create_auth_manage_page(self): """ 创建“我的-认证管理”中的更换手机号码页面 """ _pages = []
mine_mobile_path = self.register_api( "/mine_mobile/", "GET", self.mine_mobile, tenant_path=True, auth=GlobalAuth(), response=MineMobileOut )
upodate_mine_mobile_path = self.register_api( "/mine_mobile/", 'POST', self.update_mine_mobile, tenant_path=True, auth=GlobalAuth(), response=UpdateMineMobileOut )
name = '更改手机号码'
page = pages.FormPage(name=name) page.create_actions( init_action=actions.DirectAction( path=mine_mobile_path, method=actions.FrontActionMethod.GET, ), global_actions={ 'confirm': actions.ConfirmAction( path=upodate_mine_mobile_path ), } )
_pages.append(page) return _pages
复制代码
create_extension_config_schema(self)
创建插件运行时配置 schema 描述
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def create_extension_config_schema(self): """创建插件运行时配置schema描述 """ select_sms_page = pages.TablePage(select=True,name=_("指定短信插件运行时"))
self.register_front_pages(select_sms_page)
select_sms_page.create_actions( init_action=actions.DirectAction( path='/api/v1/tenants/{tenant_id}/config_select/?extension__type=sms', method=actions.FrontActionMethod.GET ) )
MobileAuthFactorSchema = create_extension_schema( 'MobileAuthFactorSchema', __file__, [ ( 'sms_config', MobileAuthFactorConfigSchema, Field( title=_('sms extension config', '短信插件运行时'), page=select_sms_page.tag, ), ), ( 'code_length', int, Field( title=_('code_length', '验证码长度'), default=6 ) ), ( 'expired', Optional[int], Field( title=_('expired', '有效期/分钟'), default=10, ) ), ], BaseAuthFactorSchema, ) self.register_auth_factor_schema(MobileAuthFactorSchema, 'mobile')
复制代码
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def create_login_page(self, event, config, config_data): """ 生成手机验证码登录页面Schema描述
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """
items = [ { "type": "text", "name":"mobile", "placeholder": "手机号码", "append": { "title": "发送验证码", "http": { "url": self.url_send_sms_code, "method": "post", "params": { "mobile": "mobile", "areacode": "86", }, }, "delay": 60 } }, { "type": "text", "name":"sms_code", "placeholder": "验证码", } ] self.add_page_form(config, self.LOGIN, "手机验证码登录", items, config_data)
复制代码
create_other_page(self, event, config, config_data)
创建其他页面（本插件无相关页面）
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def create_other_page(self, event, config, config_data): """创建其他页面（本插件无相关页面）
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """ pass
复制代码
create_password_page(self, event, config, config_data)
生成重置密码页面 Schema 描述
通过手机验证码重置密码时需提供手机号码以及对应验证码，同时此处添加新密码确认机制
注意：重置密码功能需要启用用户名密码认证插件以提供完整支持
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def create_password_page(self, event, config, config_data): """生成重置密码页面Schema描述
通过手机验证码重置密码时需提供手机号码以及对应验证码，同时此处添加新密码确认机制
注意：重置密码功能需要启用用户名密码认证插件以提供完整支持
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """ items = [ { "type": "text", "name":"mobile", "placeholder": "手机号码", "append": { "title": "发送验证码", "http": { "url": self.url_send_sms_code, "method": "post", "params": { "mobile": "mobile", "areacode": "86", }, }, } }, { "type": "text", "name":"sms_code", "placeholder": "验证码" }, { "type": "password", "name":"password", "placeholder": "密码" }, { "type": "password", "name":"checkpassword", "placeholder": "密码确认" } ] self.add_page_form(config, self.RESET_PASSWORD, "手机验证码重置密码", items, config_data)
复制代码
create_register_page(self, event, config, config_data)
生成手机验证码用户注册页面 Schema 描述
因本插件提供重置密码功能，此处需用户指定账号用户名
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def create_register_page(self, event, config, config_data): """生成手机验证码用户注册页面Schema描述
因本插件提供重置密码功能，此处需用户指定账号用户名
Args: event (Event): 事件 config (TenantExtensionConfig): 插件运行时配置 """ items = [ { "type": "text", "name": "username", "placeholder": "用户名" }, { "type": "text", "name":"mobile", "placeholder": "手机号码", "append": { "title": "发送验证码", "http": { "url": self.url_send_sms_code, "method": "post", "params": { "mobile": "mobile", "areacode": "86", }, }, "delay": 60 } }, { "type": "text", "name":"sms_code", "placeholder": "验证码" } ] self.add_page_form(config, self.REGISTER, "手机验证码注册", items, config_data)
复制代码
fix_login_page(self, event, **kwargs)
向 login_pages 填入认证元素
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/__init__.py
def fix_login_page(self, event, **kwargs): pass复制代码
load(self)
加载插件
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def load(self): """加载插件 """ super().load()
self.create_extension_config_schema()
self.register_extend_field(UserMobile, "mobile") from api.v1.schema.auth import AuthIn from api.v1.schema.user import UserCreateIn,UserItemOut,UserUpdateIn,UserListItemOut from api.v1.schema.mine import ProfileSchemaOut self.register_extend_api( AuthIn, UserCreateIn, UserItemOut, UserUpdateIn, UserListItemOut, mobile=(Optional[str],Field(title=_("电话号码"))), # areacode=(str,Field(title=_("区号"))) ) self.register_extend_api( ProfileSchemaOut, mobile=(Optional[str],Field(readonly=True)) )
# 注册发送短信接口 self.url_send_sms_code = self.register_api( '/config/{config_id}/send_sms_code/', 'POST', self.send_sms_code, tenant_path=True, auth=None, response=SendSMSCodeOut, ) print(self.url_send_sms_code)复制代码
register(self, event, **kwargs)
注册用户
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
@transaction.atomic()def register(self, event, **kwargs): """ 注册用户
Args: event (Event): 事件 """ tenant = event.tenant request = event.request data = request.POST or json.load(request.body)
mobile = data.get('mobile') sms_code = data.get('sms_code') username = data.get('username')
config = self.get_current_config(event) ret, message = self.check_mobile_exists(mobile, tenant) if not ret: return self.error(message)
if not check_sms_code(mobile, sms_code): return self.error(ErrorCode.SMS_CODE_MISMATCH)
ret, message = self.check_username_exists(username, tenant) if not ret: return self.error(message)
user = User(tenant=tenant)
user.mobile = mobile user.username = username
user.save() tenant.users.add(user) tenant.save()
return user复制代码
reset_password(self, event, **kwargs)
重置密码
Parameters:
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
def reset_password(self, event, **kwargs): """ 重置密码
Args: event (Event): 事件 """ tenant = event.tenant request = event.request data = request.POST or json.load(request.body)
mobile = data.get('mobile') sms_code = data.get('sms_code')
password = data.get('password') checkpassword = data.get('checkpassword')
if password != checkpassword: return self.error(ErrorCode.PASSWORD_IS_INCONSISTENT)
if not check_sms_code(mobile, sms_code): return self.error(ErrorCode.SMS_CODE_MISMATCH)
user = User.expand_objects.filter(tenant=tenant,mobile=mobile)
if len(user) > 1: logger.error(f'{mobile}在数据库中匹配到多个用户') return self.error(ErrorCode.CONTACT_MANAGER) if user: user = user[0] user = User.active_objects.get(id=user.get("id")) user.password = make_password(password) user.save() return self.success()
return self.error(ErrorCode.MOBILE_NOT_EXISTS_ERROR)复制代码
send_sms_code(self, request, tenant_id, config_id, data)
发送短信验证码
@operation(SendSMSCodeOut)def send_sms_code(self,request,tenant_id,config_id:str,data:SendSMSCodeIn): """发送短信验证码 """ tenant = request.tenant mobile = data.mobile config = self.get_config_by_id(config_id) if not config: return self.error(ErrorCode.CONFIG_IS_NOT_EXISTS)
if not mobile or mobile=="mobile": return self.error(ErrorCode.MOBILE_EMPTY)
code = create_sms_code(tenant,mobile,config.config.get('code_length',6),config.config.get("expired",10)*60)
responses = dispatch_event( Event( tag=SEND_SMS, tenant=tenant, request=request, data={ "config_id":config.config["sms_config"]["id"], "mobile":data.mobile, "code": code, "areacode": data.areacode, "username": request.user.username if request.user else "" }, packages=config.config["sms_config"]["package"]
) )
if not responses: return self.error(ErrorCode.SMS_EXTENSION_NOT_EXISTS) useless, (data, extension) = responses[0] if data: return self.success() else: return self.error(ErrorCode.SMS_SEND_FAILED)
复制代码
update_mine_mobile(self, request, tenant_id, data)
普通用户：更新手机号码
Source code in extension_root/com_longgui_auth_factor_mobile/init.py
@operation(UpdateMineMobileOut,roles=[TENANT_ADMIN, PLATFORM_ADMIN, NORMAL_USER])def update_mine_mobile(self, request, tenant_id: str,data:UpdateMineMobileIn): """ 普通用户：更新手机号码 """ mobile = data.mobile ret, message = self.check_mobile_exists(mobile, request.tenant) if not ret: return self.error(message)
if not check_sms_code(request.tenant,mobile,data.code): return self.error(ErrorCode.SMS_CODE_MISMATCH)
user = request.user user.mobile=data.mobile user.save()
return self.success()复制代码
extension_root.com_longgui_auth_factor_mobile.sms
check_sms_code(tenant, mobile, code)
验证短信验证码
def check_sms_code(tenant,mobile, code): """ 验证短信验证码 """ c_code = cache.get(tenant,gen_sms_code_key(mobile)) return c_code == code复制代码
create_sms_code(tenant, phone_number, uth_code_length=6, expired=None)
生成短信验证码并存储至缓存
Source code in extension_root/com_longgui_auth_factor_mobile/sms.py
def create_sms_code(tenant,phone_number,uth_code_length=6,expired:Optional[int]=None): """生成短信验证码并存储至缓存 """ code = gen_sms_code(uth_code_length) cache.set(tenant,gen_sms_code_key(phone_number), code, expired=expired) return code复制代码
gen_sms_code_key(mobile)
生成短信验证码的 key
Source code in extension_root/com_longgui_auth_factor_mobile/sms.py
def gen_sms_code_key(mobile): ''' 生成短信验证码的key ''' return f'sms:{mobile}'复制代码
划线
评论
复制
发布于: 刚刚阅读数: 4
版权声明: 本文为 InfoQ 作者【龙归科技】的原创文章。
原文链接:【http://xie.infoq.cn/article/a3a84358b97cab72f658eae2d】。
本文遵守【CC-BY 4.0】协议，转载请保留原文出处及本版权声明。
龙归科技
关注
还未添加个人签名 2021.01.11 加入
「龙归科技」致力于让每个组织拥有专属的自动化办公操作系统，助力企业或政府拥抱（Cloud Native First）云原生优先战略，帮助客户构筑以「身份与应用」为中心的现代化 IT 基础设施！
评论