Coverage for model/auth.py: 98%

1# Standard library

2from random import SystemRandom

3from typing import Optional

4import csv

5import io

6import logging

7# Related third party imports

8from fastapi import APIRouter, Cookie, Depends, Request

9from .utils.request import get_ip

10# Local application

11from mongo import *

12from mongo import engine

13from mongo.utils import hash_id

14from .utils import *

15from .schemas import (

16 LoginBody,

17 SignupBody,

18 ChangePasswordBody,

19 CheckUsernameBody,

20 CheckEmailBody,

21 ResendEmailBody,

22 ActivateUserBody,

23 PasswordRecoveryBody,

24 AuthAddUserBody,

25 BatchSignupBody,

26 GetMeQuery,

27)

29import string

31__all__ = (

32 'auth_router',

33 'login_required',

34 'identity_verify',

35 'get_verify_link',

36)

38logger = logging.getLogger(__name__)

40auth_router = APIRouter()

42VERIFY_TEXT = '''\

43Welcome! you've signed up successfully!

44Enter Normal OJ to active your account via this link:

45{url}

46'''

48VERIFY_HTML = '''\

49<!DOCTYPE html><html lang="en"><head><title>template</title><meta http-equiv="Content-Type" content="text/html; charset=utf-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><meta name="viewport" content="width=device-width, initial-scale=1.0 "><meta name="format-detection" content="telephone=no"><link href="https://fonts.googleapis.com/css?family=Lato:300,400,600,700,800" rel="stylesheet"><style>.em_body {{margin: 0px;padding: 0px;background-color: #efefef;}}.em_full_wrap {{vertical-align: top;width: 100%;border-spacing: 0px;border-collapse: separate;border: 0px;background-color: #efefef;margin-left: auto; margin-right: auto;}}.em_main_table {{width: 700px;border-spacing: 0px;border-collapse: separate;align-self: center;margin-left:auto; margin-right:auto;}}.em_full_wrap td, .em_main_table td {{padding: 0px;vertical-align: top;text-align: center;}}</style></head><body class="em_body"><table class="em_full_wrap"><tbody><tr><td><table class="em_main_table"><tr><td style="padding:35px 70px 30px; background-color: #003865"><table style="width: 100%; border-spacing: 0px; border-collapse: separate; border: 0px; margin-left: auto; margin-right: auto;"><tbody><tr><td style="font-family:'Lato', Arial, sans-serif; font-size:16px; line-height:30px; color:#fff; vertical-align: top; text-align: center;">Normal Online Judge Email Verification</td></tr><tr><td><hr></td></tr><tr><td style="font-family:'Lato', Arial, sans-serif; font-size:20px; line-height:22px; color:#fff; padding:12px; vertical-align: top; text-align: center;">Welcome! you've signed up successfully!<br><br>Enter Normal OJ to active your account via this link.</td></tr><tr><td class="em_h20" style="font-size:0px; line-height:0px; height:25px;"> </td></tr><tr><td style="vertical-align: top; text-align: center;"><form target="_blank" action="{url}"><button type="submit" style="background:#A6DAEF; border-color: #fff; border-radius: 5px; font-family:'Lato', Arial, sans-serif; font-size:16px; line-height:22px; box-shadow: 0 8px 16px 0 rgba(0,0,0,0.2), 0 6px 20px 0 rgba(0,0,0,0.19); cursor: pointer;">Active Account</button></form></td></tr></tbody></table></td></tr><tr><td style="padding:18px 30px; background-color: #f6f7f8"><table style="width: 100%; border-spacing: 0px; border-collapse: separate; border: 0px; margin-left: auto; margin-right: auto;"><tbody><tr><td style="font-family:'Lato', Arial, sans-serif; font-size:11px; line-height:18px; color:#999999; vertical-align: top; text-align: center;">© 2020 Normal Online Judge. All Rights Reserved.</td></tr></tbody></table></td></tr></table></td></tr></tbody></table></body></html>

50'''

53def login_required(piann: str | None = Cookie(default=None)) -> User:

54 '''Check if the user is logged in.

56 Raises:

57 - 403 Not Logged In

58 - 403 Invalid Token

59 - 403 Authorization Expired

60 - 403 Inactive User

61 '''

62 if piann is None:

63 raise NOJException('Not Logged In', 403)

64 json = jwt_decode(piann)

65 if json is None or not json.get('secret'):

66 raise NOJException('Invalid Token', 403)

67 user = User(json['data']['username'])

68 if json['data'].get('userId') != user.user_id:

69 raise NOJException('Authorization Expired', 403)

70 if not user.active:

71 raise NOJException('Inactive User', 403)

72 return user

75def identity_verify(*roles):

76 '''Verify a logged-in user's identity against allowed roles.'''

78 def dependency(user: User = Depends(login_required)) -> User:

79 if user.role not in roles:

80 raise NOJException('Insufficient Permissions', 403)

81 return user

83 return Depends(dependency)

86def get_verify_link(user: User, request: Request) -> str:

87 return str(request.url_for('active_redirect', token=user.cookie))

90@auth_router.get('/session')

91def logout():

92 '''Logout a user.'''

93 cookies = {'jwt': None, 'piann': None}

94 return HTTPResponse('Goodbye', cookies=cookies)

97@auth_router.post('/session')

98def login(body: LoginBody, ip_addr: str = Depends(get_ip)):

99 '''Login a user.'''

100 try:

101 user = User.login(body.username, body.password, ip_addr)

102 except DoesNotExist:

103 return HTTPError('Login Failed', 403)

104 if not user.active:

105 return HTTPError('Invalid User', 403)

106 cookies = {'piann_httponly': user.secret, 'jwt': user.cookie}

107 return HTTPResponse('Login Success', cookies=cookies)

108

109

110@auth_router.post('/signup')

111def signup(body: SignupBody, request: Request):

112 try:

113 user = User.signup(body.username, body.password, body.email)

114 except ValidationError as ve:

115 return HTTPError('Signup Failed', 400, data=ve.to_dict())

116 except NotUniqueError:

117 return HTTPError('User Exists', 400)

118 except ValueError:

119 return HTTPError('Not Allowed Name', 400)

120 verify_link = get_verify_link(user, request)

121 text = VERIFY_TEXT.format(url=verify_link)

122 html = VERIFY_HTML.format(url=verify_link)

123 send_noreply([body.email], '[N-OJ] Varify Your Email', text, html)

124 return HTTPResponse('Signup Success')

125

126

127@auth_router.post('/change-password')

128def change_password(

129 body: ChangePasswordBody,

130 user: User = Depends(login_required),

131 ip_addr: str = Depends(get_ip),

132):

133 try:

134 User.login(user.username, body.old_password, ip_addr)

135 except DoesNotExist:

136 return HTTPError('Wrong Password', 403)

137 user.change_password(body.new_password)

138 cookies = {'piann_httponly': user.secret}

139 return HTTPResponse('Password Has Been Changed', cookies=cookies)

140

141

142@auth_router.post('/check/username')

143def check_username(body: CheckUsernameBody):

144 try:

145 User.get_by_username(body.username)

146 except DoesNotExist:

147 return HTTPResponse('Username Can Be Used', data={'valid': 1})

148 return HTTPResponse('User Exists', data={'valid': 0})

149

150

151@auth_router.post('/check/email')

152def check_email(body: CheckEmailBody):

153 try:

154 User.get_by_email(body.email)

155 except DoesNotExist:

156 return HTTPResponse('Email Can Be Used', data={'valid': 1})

157 return HTTPResponse('Email Has Been Used', data={'valid': 0})

158

159

160@auth_router.post('/check/{item}')

161def check_invalid(item: str):

162 return HTTPError('Ivalid Checking Type', 400)

163

164

165@auth_router.post('/resend-email')

166def resend_email(body: ResendEmailBody, request: Request):

167 try:

168 user = User.get_by_email(body.email)

169 except DoesNotExist:

170 return HTTPError('User Not Exists', 400)

171 if user.active:

172 return HTTPError('User Has Been Actived', 400)

173 verify_link = get_verify_link(user, request)

174 send_noreply([body.email], '[N-OJ] Varify Your Email', verify_link)

175 return HTTPResponse('Email Has Been Resent')

176

177

178@auth_router.get('/active/{token}', name='active_redirect')

179def active_redirect(token: str):

180 '''Redirect user to active page.'''

181 json = jwt_decode(token)

182 if json is None:

183 return HTTPError('Invalid Token', 403)

184 user = User(json['data']['username'])

185 cookies = {'piann_httponly': user.secret, 'jwt': user.cookie}

186 return HTTPRedirect('/email_verify', cookies=cookies)

187

188

189@auth_router.post('/active')

190def activate_user(

191 body: ActivateUserBody,

192 piann: str | None = Cookie(default=None),

193):

194 '''User: active: false -> true'''

195 if body.agreement is not True:

196 return HTTPError('Not Confirm the Agreement', 403)

197 json = jwt_decode(piann)

198 if json is None or not json.get('secret'):

199 return HTTPError('Invalid Token.', 403)

200 user = User(json['data']['username'])

201 if not user:

202 return HTTPError('User Not Exists', 400)

203 if user.active:

204 return HTTPError('User Has Been Actived', 400)

205 try:

206 user.activate(body.profile)

207 except engine.DoesNotExist as e:

208 return HTTPError(str(e), 404)

209 cookies = {'jwt': user.cookie}

210 return HTTPResponse('User Is Now Active', cookies=cookies)

211

212

213@auth_router.post('/password-recovery')

214def password_recovery(body: PasswordRecoveryBody):

215 email = body.email

216 try:

217 user = User.get_by_email(email)

218 except DoesNotExist:

219 return HTTPError('User Not Exists', 400)

220 new_password = (lambda r: ''.join(

221 r.choice(string.hexdigits)

222 for i in range(r.randint(12, 24))))(SystemRandom())

223 user_id2 = hash_id(user.username, new_password)

224 user.update(user_id2=user_id2)

225 send_noreply(

226 [email], '[N-OJ] Password Recovery',

227 f'Your alternative password is {new_password}.\nPlease login and change your password.'

228 )

229 return HTTPResponse('Recovery Email Has Been Sent')

230

231

232@auth_router.post('/user')

233def add_user(

234 body: AuthAddUserBody,

235 user: User = identity_verify(0),

236):

237 '''Directly add a user without activation. Admin only.'''

238 try:

239 User.signup(

240 body.username,

241 body.password,

242 body.email,

243 ).activate()

244 except ValidationError as ve:

245 return HTTPError('Signup Failed', 400, data=ve.to_dict())

246 except NotUniqueError:

247 return HTTPError('User Exists', 400)

248 except ValueError:

249 return HTTPError('Not Allowed Name', 400)

250 return HTTPResponse()

251

252

253@auth_router.post('/batch-signup')

254def batch_signup(

255 body: BatchSignupBody,

256 user: User = identity_verify(0),

257):

258 course = None

259 if body.course is not None:

260 try:

261 course = Course(body.course)

262 if not course:

263 return HTTPError(f'Course not found', 404)

264 except engine.DoesNotExist as e:

265 return HTTPError(str(e), 404)

266 try:

267 new_users = [*csv.DictReader(io.StringIO(body.new_users))]

268 except csv.Error as e:

269 logger.info(f'Error parse csv file [err={e}]')

270 return HTTPError('Invalid file content', 400)

271 force = body.force if body.force is not None else False

272 try:

273 new_users = User.batch_signup(

274 new_users=new_users,

275 course=course,

276 force=force,

277 )

278 except ValueError as e:

279 return HTTPError(str(e), 400)

280 return HTTPResponse()

281

282

283@auth_router.get('/me')

284def get_me(

285 query: GetMeQuery = Depends(),

286 user: User = Depends(login_required),

287):

288 fields = query.fields

289 default = [

290 'username',

291 'email',

292 'md5',

293 'active',

294 'role',

295 'editorConfig',

296 'displayedName',

297 'bio',

298 ]

299 if fields is None:

300 fields = default

301 else:

302 fields = fields.split(',')

303 try:

304 return HTTPResponse(data=user.properties(*fields))

305 except ValueError as e:

306 return HTTPError(str(e), 400)