Coverage for apps/borrower_portal/api.py: 98%

1"""Borrower account endpoints that need no session: register, password reset, one-time codes 

2(mounted at "" beside the shared ``local_auth_api`` routes; add-borrower-portal-api).""" 

3 

4from __future__ import annotations 

5 

6import secrets 

7 

8from django.conf import settings 

9from django.db import IntegrityError, transaction 

10from django.utils import timezone 

11from ninja import Router 

12 

13from apps.borrower_portal.models import AuthToken, BorrowerAccount 

14from apps.borrower_portal.schemas import ( 

15 BorrowerAuthResultOut, 

16 OtpRequestInput, 

17 OtpRequestOut, 

18 OtpVerifyInput, 

19 RegisterInput, 

20 ResetInput, 

21) 

22from apps.borrower_portal.services import ( 

23 RESET_TTL_MINUTES, 

24 borrower_session, 

25 company_name, 

26 hash_token, 

27 issue_auth_token, 

28 portal_url, 

29 send_borrower_message, 

30) 

31from apps.communications import emails 

32from domain.pos import PASSWORD_MSG, first_error, is_strong_password, validate_register 

33from domain.result import AppError, DomainError 

34 

35router = Router(tags=["borrower-auth"]) 

36 

37EMAIL_TAKEN = "An account with that email already exists." 

38STAFF_MIN_PASSWORD = 4 

39 

40 

41@router.post("/auth/register", response={201: BorrowerAuthResultOut}) 

42def register(request, payload: RegisterInput): 

43 problem = first_error(validate_register(payload.name, payload.email, payload.password)) 

44 if problem: 

45 raise DomainError([problem]) 

46 email = payload.email.strip().lower() 

47 if BorrowerAccount.objects.filter(email=email).exists(): 

48 raise DomainError([AppError("VALIDATION", EMAIL_TAKEN, "email")]) 

49 account = BorrowerAccount(email=email, name=payload.name.strip()) 

50 account.set_password(payload.password) 

51 try: 

52 with transaction.atomic(): 

53 account.save() 

54 except IntegrityError as exc: # a concurrent registration won the race 

55 raise DomainError([AppError("VALIDATION", EMAIL_TAKEN, "email")]) from exc 

56 return 201, borrower_session(account) 

57 

58 

59# --- password reset (borrower-auth: password reset; staff-auth delta) ------------------------- 

60def send_reset_link(email: str) -> str | None: 

61 """``POST /api/auth/forgot`` for a borrower account: email a one-hour reset link to the 

62 borrower portal. Returns the link (shown as ``devResetLink`` in development), or None when 

63 the email has no active borrower account — staff accounts get nothing (design.md D10).""" 

64 account = BorrowerAccount.objects.filter( 

65 email=(email or "").strip().lower(), is_active=True 

66 ).first() 

67 if account is None: 

68 return None 

69 raw = issue_auth_token(account, AuthToken.KIND_RESET) 

70 link = portal_url(f"/reset-password?token={raw}") 

71 subject, content = emails.password_reset(company_name(), link) 

72 send_borrower_message( 

73 channel="email", to=account.email, subject=subject, content=content, trigger="auth:reset" 

74 ) 

75 return link 

76 

77 

78def send_staff_reset_link(email: str) -> str | None: 

79 """``POST /api/auth/forgot`` for a staff user: email a one-hour reset link to the Staff LOS 

80 (add-sendgrid-email). Returns the link (``devResetLink`` in development) or None when the email 

81 has no active staff user.""" 

82 from datetime import timedelta 

83 

84 from apps.accounts.models import User 

85 from apps.communications.delivery import deliver_email 

86 

87 user = User.objects.filter(email__iexact=(email or "").strip(), is_active=True).first() 

88 if user is None: 

89 return None 

90 now = timezone.now() 

91 AuthToken.objects.filter( 

92 staff_user=user, kind=AuthToken.KIND_RESET, used_at__isnull=True 

93 ).update(used_at=now) 

94 raw = secrets.token_urlsafe(32) 

95 AuthToken.objects.create( 

96 staff_user=user, 

97 kind=AuthToken.KIND_RESET, 

98 token_hash=hash_token(raw), 

99 expires_at=now + timedelta(minutes=RESET_TTL_MINUTES), 

100 ) 

101 link = f"{settings.STAFF_LOS_URL}/reset-password?token={raw}" 

102 subject, content = emails.password_reset(company_name(), link) 

103 deliver_email( 

104 to=user.email, 

105 subject=subject, 

106 content=content, 

107 trigger="auth:reset", 

108 automated=True, 

109 track_clicks=False, 

110 ) 

111 return link 

112 

113 

114def _usable_token(raw: str, kind: str, account: BorrowerAccount | None = None): 

115 """The token row for ``raw``: (row, None) when usable, else (row-or-None, error code).""" 

116 query = AuthToken.objects.filter(token_hash=hash_token(raw or ""), kind=kind) 

117 if account is not None: 

118 query = query.filter(account=account) 

119 token = query.select_related("account", "staff_user").order_by("-created_at").first() 

120 owner = (token.account or token.staff_user) if token is not None else None 

121 if token is None or token.used_at is not None or owner is None or not owner.is_active: 

122 return token, "invalid" 

123 if token.expires_at <= timezone.now(): 

124 return token, "expired" 

125 return token, None 

126 

127 

128@router.post("/auth/reset", response={204: None}) 

129def reset_password(request, payload: ResetInput): 

130 token, problem = _usable_token(payload.token, AuthToken.KIND_RESET) 

131 if problem == "invalid": 

132 raise DomainError( 

133 [AppError("RESET_INVALID", "This reset link is invalid or has already been used.")] 

134 ) 

135 if problem == "expired": 

136 raise DomainError( 

137 [AppError("RESET_EXPIRED", "This reset link has expired. Request a new one.")] 

138 ) 

139 if token.staff_user is not None: 

140 # Staff keep the staff password rule (same as Staff LOS My account → Change password). 

141 if len(payload.password) < STAFF_MIN_PASSWORD: 

142 raise DomainError( 

143 [AppError("VALIDATION", "New password must be at least 4 characters.", "password")] 

144 ) 

145 elif not is_strong_password(payload.password): 

146 raise DomainError([AppError("VALIDATION", PASSWORD_MSG, "password")]) 

147 with transaction.atomic(): 

148 owner = token.account or token.staff_user 

149 owner.set_password(payload.password) 

150 owner.save(update_fields=["password"] if token.staff_user else ["password", "updated_at"]) 

151 token.used_at = timezone.now() 

152 token.save(update_fields=["used_at"]) 

153 return 204, None 

154 

155 

156# --- one-time sign-in codes (borrower-auth: one-time sign-in code) ---------------------------- 

157@router.post("/auth/otp/request", response=OtpRequestOut, exclude_none=True) 

158def request_otp(request, payload: OtpRequestInput): 

159 account = BorrowerAccount.objects.filter( 

160 email=(payload.email or "").strip().lower(), is_active=True 

161 ).first() 

162 if account is None: 

163 return {"sent": True} # same answer whether or not the account exists 

164 code = issue_auth_token(account, AuthToken.KIND_OTP) 

165 subject, content = emails.sign_in_code(company_name(), code) 

166 # Always by email: there is no SMS provider yet (add-sendgrid-email). 

167 send_borrower_message( 

168 channel="email", to=account.email, subject=subject, content=content, trigger="auth:otp" 

169 ) 

170 return {"sent": True, "devOtp": code if settings.DEBUG else None} 

171 

172 

173@router.post("/auth/otp/verify", response=BorrowerAuthResultOut) 

174def verify_otp(request, payload: OtpVerifyInput): 

175 account = BorrowerAccount.objects.filter( 

176 email=(payload.email or "").strip().lower(), is_active=True 

177 ).first() 

178 token, problem = ( 

179 _usable_token(payload.code, AuthToken.KIND_OTP, account) if account else (None, "invalid") 

180 ) 

181 if problem: 

182 raise DomainError([AppError("MFA_INVALID", "That code is wrong or has expired.", "code")]) 

183 assert token is not None # _usable_token returns a token whenever problem is falsy 

184 token.used_at = timezone.now() 

185 token.save(update_fields=["used_at"]) 

186 return borrower_session(account)