Coverage for app/venv/lib/python3.14/site-packages/weblate/accounts/pipeline.py: 14%

3# SPDX-License-Identifier: GPL-3.0-or-later

4from __future__ import annotations

6import re

7import time

8import unicodedata

10from django.conf import settings

11from django.contrib.auth.models import AnonymousUser

12from django.http import HttpResponseRedirect

13from django.shortcuts import redirect

14from django.urls import reverse

15from django.utils.http import url_has_allowed_host_and_scheme, urlencode

16from django.utils.translation import gettext

17from django_otp import DEVICE_ID_SESSION_KEY

18from social_core.exceptions import AuthAlreadyAssociated, AuthMissingParameter

19from social_core.pipeline.partial import partial

20from social_core.utils import PARTIAL_TOKEN_SESSION_NAME

22from weblate.accounts.models import AuditLog, VerifiedEmail

23from weblate.accounts.notifications import send_notification_email

24from weblate.accounts.templatetags.authnames import get_auth_name

25from weblate.accounts.utils import (

26 SESSION_SECOND_FACTOR_SOCIAL,

27 SESSION_SECOND_FACTOR_USER,

28 adjust_session_expiry,

29 cycle_session_keys,

30 invalidate_reset_codes,

31)

32from weblate.auth.models import Invitation, User, get_anonymous

33from weblate.trans.defines import FULLNAME_LENGTH

34from weblate.utils import messages

35from weblate.utils.ratelimit import reset_rate_limit

36from weblate.utils.requests import request

37from weblate.utils.validators import (

38 CRUD_RE,

39 USERNAME_MATCHER,

40 EmailValidator,

41 clean_fullname,

42)

44STRIP_MATCHER = re.compile(r"[^\w\s.@+-]")

45CLEANUP_MATCHER = re.compile(r"[-\s]+")

48class UsernameAlreadyAssociated(AuthAlreadyAssociated):

49 pass

52class EmailAlreadyAssociated(AuthAlreadyAssociated):

53 pass

56def get_github_emails(access_token):

57 """Get real e-mail from GitHub."""

58 response = request(

59 "get",

60 "https://api.github.com/user/emails",

61 headers={"Authorization": f"token {access_token}"},

62 timeout=10.0,

63 )

64 data = response.json()

65 email = None

66 primary = None

67 public = None

68 emails = []

69 for entry in data:

70 # Skip noreply e-mail only if we need deliverable e-mails

71 if entry["email"].endswith("@users.noreply.github.com"):

72 # Add E-Mail and set is_deliverable to false

73 emails.append((entry["email"], False))

74 continue

75 # Skip not verified ones

76 if not entry["verified"]:

77 continue

79 # Add E-Mail and set is_deliverable to true

80 emails.append((entry["email"], True))

81 if entry.get("visibility") == "public":

82 # There is just one public mail, prefer it

83 public = entry["email"]

84 continue

85 email = entry["email"]

86 if entry["primary"]:

87 primary = entry["email"]

88 return public or primary or email, emails

91@partial

92def reauthenticate(

93 strategy, backend, user: User, social, uid, weblate_action, **kwargs

94):

95 """Force authentication when adding new association."""

96 session = strategy.request.session

97 if session.pop("reauthenticate_done", False):

98 return None

99 if weblate_action != "activation":

100 return None

101 if user and not social and user.has_usable_password():

102 session["reauthenticate"] = {

103 "backend": backend.name,

104 "backend_verbose": str(get_auth_name(backend.name)),

105 "uid": uid,

106 "user_pk": user.pk,

107 }

108 return redirect("confirm")

109 return None

110

111

112@partial

113def require_email(backend, details, weblate_action, user=None, is_new=False, **kwargs):

114 """Force entering e-mail for backends which don't provide it."""

115 if backend.name == "github":

116 email, emails = get_github_emails(kwargs["response"]["access_token"])

117 details["verified_emails"] = emails

118 if email is not None:

119 details["email"] = email

120

121 # Remove any pending e-mail validation codes

122 if details.get("email") and backend.name == "email":

123 invalidate_reset_codes(emails=(details["email"],))

124 # Remove all account reset codes

125 if user and weblate_action == "reset":

126 invalidate_reset_codes(user=user)

127

128 if user and user.email:

129 # Force validation of new e-mail address

130 if backend.name == "email":

131 return {"is_new": True}

132

133 return None

134

135 if is_new and not details.get("email"):

136 raise AuthMissingParameter(backend, "email")

137 return None

138

139

140def send_validation(strategy, backend, code, partial_token) -> None:

141 """Send verification e-mail."""

142 # We need to have existing session

143 session = strategy.request.session

144 if not session.session_key:

145 session.create()

146 session["registration-email-sent"] = True

147

148 url = "{}?verification_code={}&partial_token={}".format(

149 reverse("social:complete", args=(backend.name,)), code.code, partial_token

150 )

151

152 context = {"url": url, "validity": settings.AUTH_TOKEN_VALID // 3600}

153

154 template = "activation"

155 if session.get("password_reset"):

156 template = "reset"

157 elif session.get("account_remove"):

158 template = "remove"

159

160 # Use None for user to enable linking by e-mail later for the registration

161 AuditLog.objects.create(

162 strategy.request.user if strategy.request.user.is_authenticated else None,

163 strategy.request,

164 "sent-email",

165 email=code.email,

166 )

167

168 # Send actual confirmation

169 send_notification_email(None, [code.email], template, info=url, context=context)

170

171

172@partial

173def password_reset(

174 strategy,

175 backend,

176 user: User,

177 social,

178 details,

179 weblate_action,

180 current_partial,

181 **kwargs,

182):

183 """Set unusable password on reset."""

184 if strategy.request is not None and user is not None and weblate_action == "reset":

185 AuditLog.objects.create(

186 user,

187 strategy.request,

188 "reset",

189 method=backend.name,

190 name=social.uid,

191 password=user.password,

192 )

193 user.set_unusable_password()

194 user.save(update_fields=["password"])

195 # Remove partial pipeline, we do not need it

196 strategy.really_clean_partial_pipeline(current_partial.token)

197 session = strategy.request.session

198 # Store user ID

199 session["perform_reset"] = user.pk

200 # Redirect to form to change password

201 return redirect("password_reset")

202 return None

203

204

205@partial

206def remove_account(

207 strategy,

208 backend,

209 user: User,

210 social,

211 details,

212 weblate_action: str,

213 current_partial,

214 **kwargs,

215):

216 """Set unusable password on reset."""

217 if strategy.request is not None and user is not None and weblate_action == "remove":

218 # Remove partial pipeline, we do not need it

219 strategy.really_clean_partial_pipeline(current_partial.token)

220 # Set short session expiry

221 session = strategy.request.session

222 session["remove_confirm"] = True

223 # Reset rate limit to allow form submission

224 reset_rate_limit("remove", strategy.request)

225 # Redirect to the confirmation form

226 return redirect("remove")

227 return None

228

229

230def verify_open(

231 strategy,

232 backend,

233 user: User,

234 weblate_action: str,

235 invitation_link: Invitation | None,

236 **kwargs,

237) -> None:

238 """Check whether it is possible to create new user."""

239 # Ensure it's still same user (if sessions was kept as this is to avoid

240 # completing authentication under different user than initiated it, with

241 # new session, it will complete as new user)

242 current_user = strategy.request.user.pk

243 init_user = strategy.request.session.get("social_auth_user")

244 if strategy.request.session.session_key and current_user != init_user:

245 raise AuthMissingParameter(backend, "user")

246

247 # Check whether registration is open

248 if (

249 not user

250 and weblate_action not in {"reset", "remove"}

251 and not invitation_link

252 and (not settings.REGISTRATION_OPEN or settings.REGISTRATION_ALLOW_BACKENDS)

253 and backend.name not in settings.REGISTRATION_ALLOW_BACKENDS

254 ):

255 raise AuthMissingParameter(backend, "disabled")

256

257

258def cleanup_next(strategy, **kwargs):

259 # This is mostly fix for lack of next validation in Python Social Auth

260 # see https://github.com/python-social-auth/social-core/issues/62

261 url = strategy.session_get("next")

262 if url and not url_has_allowed_host_and_scheme(url, allowed_hosts=None):

263 strategy.session_set("next", None)

264 if url_has_allowed_host_and_scheme(kwargs.get("next", ""), allowed_hosts=None):

265 return None

266 return {"next": None}

267

268

269def store_params(strategy, user: User, **kwargs):

270 """Store Weblate specific parameters in the pipeline."""

271 # Map standard Django anonymuos user to Weblate database backed one

272 if isinstance(strategy.request.user, AnonymousUser):

273 strategy.request.user = get_anonymous()

274 # Registering user

275 registering_user = user.pk if user and user.is_authenticated else None

276

277 # Pipeline action

278 session = strategy.request.session

279 if session.get("password_reset"):

280 action = "reset"

281 elif session.get("account_remove"):

282 action = "remove"

283 else:

284 action = "activation"

285

286 invitation = None

287 if invitation_pk := session.get("invitation_link"):

288 try:

289 invitation = Invitation.objects.get(pk=invitation_pk)

290 except Invitation.DoesNotExist:

291 del session["invitation_link"]

292 invitation_pk = None

293

294 return {

295 "weblate_action": action,

296 "registering_user": registering_user,

297 "weblate_expires": int(time.time() + settings.AUTH_TOKEN_VALID),

298 "invitation_link": invitation,

299 "invitation_pk": str(invitation_pk) if invitation_pk else None,

300 }

301

302

303def verify_username(strategy, backend, details, username, user=None, **kwargs) -> None:

304 """

305 Verify whether username is still free.

306

307 It can happen that user has registered several times or other user has taken the

308 username meanwhile.

309 """

310 if user or not username:

311 return

312 if User.objects.filter(username=username).exists():

313 raise UsernameAlreadyAssociated(backend, "Username exists")

314 return

315

316

317def revoke_mail_code(strategy, details, **kwargs) -> None:

318 """

319 Remove old mail validation code for Python Social Auth.

320

321 PSA keeps them around, but we really don't need them again.

322 """

323 data = strategy.request_data()

324 if "email" in details and details["email"] and "verification_code" in data:

325 try:

326 code = strategy.storage.code.objects.get(

327 code=data["verification_code"], email=details["email"], verified=True

328 )

329 code.delete()

330 except strategy.storage.code.DoesNotExist:

331 return

332

333

334def ensure_valid(

335 strategy,

336 backend,

337 user: User,

338 registering_user,

339 weblate_action,

340 weblate_expires,

341 new_association,

342 details,

343 **kwargs,

344) -> None:

345 """Ensure the activation link is still."""

346 # Didn't the link expire?

347 if weblate_expires < time.time():

348 raise AuthMissingParameter(backend, "expires")

349

350 # We allow password reset for unauthenticated users

351 if weblate_action == "reset":

352 if strategy.request.user.is_authenticated:

353 messages.warning(

354 strategy.request,

355 gettext("You can not complete password reset while signed in."),

356 )

357 messages.warning(

358 strategy.request, gettext("The registration link has been invalidated.")

359 )

360 raise AuthMissingParameter(backend, "user")

361 return

362

363 # Add e-mail/register should stay on same user

364 current_user = user.pk if user and user.is_authenticated else None

365

366 if current_user != registering_user:

367 if registering_user is None:

368 messages.warning(

369 strategy.request,

370 gettext("You can not complete registration while signed in."),

371 )

372 else:

373 messages.warning(

374 strategy.request,

375 gettext("You can confirm your registration only while signed in."),

376 )

377 messages.warning(

378 strategy.request, gettext("The registration link has been invalidated.")

379 )

380

381 raise AuthMissingParameter(backend, "user")

382

383 # Verify if this mail is not used on other accounts

384 if new_association:

385 if "email" not in details:

386 raise AuthMissingParameter(backend, "email")

387 same = VerifiedEmail.objects.filter(email__iexact=details["email"])

388 if user:

389 same = same.exclude(social__user=user)

390

391 if not settings.REGISTRATION_REBIND and same.exists():

392 AuditLog.objects.create(same[0].social.user, strategy.request, "connect")

393 raise EmailAlreadyAssociated(backend, "E-mail exists")

394

395 validator = EmailValidator()

396 # This raises ValidationError

397 validator(details["email"])

398

399

400def store_email(strategy, backend, user: User, social, details, **kwargs) -> None:

401 """Store verified e-mail."""

402 # The email can be empty for some services

403 if details.get("verified_emails"):

404 # For some reasons tuples get converted to lists inside python social auth

405 current = {tuple(verified) for verified in details["verified_emails"]}

406 existing = set(social.verifiedemail_set.values_list("email", "is_deliverable"))

407 for remove in existing - current:

408 social.verifiedemail_set.filter(

409 email=remove[0], is_deliverable=remove[1]

410 ).delete()

411 for add in current - existing:

412 social.verifiedemail_set.create(email=add[0], is_deliverable=add[1])

413 elif details.get("email"):

414 verified, created = VerifiedEmail.objects.get_or_create(

415 social=social, defaults={"email": details["email"]}

416 )

417 if (

418 not created and verified.email != details["email"]

419 ) or not verified.is_deliverable:

420 verified.email = details["email"]

421 verified.is_deliverable = True

422 verified.save()

423

424

425def handle_invite(

426 strategy, backend, user: User, social, invitation_pk: str, **kwargs

427) -> None:

428 # Accept triggering invitation

429 if invitation_pk:

430 Invitation.objects.get(pk=invitation_pk).accept(strategy.request, user)

431 # Merge possibly pending invitations for this e-mail address

432 Invitation.objects.filter(email=user.email).update(user=user, email="")

433

434

435def notify_connect(

436 strategy,

437 details,

438 backend,

439 user: User,

440 social,

441 weblate_action,

442 new_association=False,

443 is_new=False,

444 **kwargs,

445) -> None:

446 """Notify about adding new link."""

447 # Adjust possibly pending email confirmation audit logs

448 AuditLog.objects.filter(

449 user=None,

450 activity="sent-email",

451 params={"email": details["email"]},

452 ).update(user=user)

453 if user and not is_new and weblate_action != "reset":

454 if new_association:

455 action = "auth-connect"

456 else:

457 action = "login"

458 adjust_session_expiry(request=strategy.request, user=user)

459 AuditLog.objects.create(

460 user,

461 strategy.request,

462 action,

463 method=backend.name,

464 name=social.uid,

465 )

466 # Remove partial pipeline

467 session = strategy.request.session

468 if PARTIAL_TOKEN_SESSION_NAME in session:

469 strategy.really_clean_partial_pipeline(session[PARTIAL_TOKEN_SESSION_NAME])

470

471

472def user_full_name(strategy, details, username, user=None, **kwargs) -> None:

473 """Update user full name using data from provider."""

474 if user and not user.full_name:

475 full_name = details.get("fullname") or ""

476 full_name = full_name.strip()

477

478 if not full_name and ("first_name" in details or "last_name" in details):

479 first_name = details.get("first_name") or ""

480 last_name = details.get("last_name") or ""

481

482 if first_name and first_name not in last_name:

483 full_name = f"{first_name} {last_name}"

484 elif first_name:

485 full_name = first_name

486 else:

487 full_name = last_name

488

489 if CRUD_RE.match(full_name):

490 full_name = ""

491

492 if not full_name and username:

493 full_name = username

494

495 if not full_name and user.username:

496 full_name = user.username

497

498 full_name = clean_fullname(full_name)

499

500 # The User model limit is 150 chars

501 if len(full_name) > FULLNAME_LENGTH:

502 full_name = full_name[:FULLNAME_LENGTH]

503

504 if full_name:

505 user.full_name = full_name

506 strategy.storage.user.changed(user)

507

508

509def slugify_username(value):

510 """

511 Clean up username.

512

513 This is based on Django slugify with exception of lowercasing

514

515 - Converts to ascii

516 - Removes not wanted chars

517 - Merges whitespaces and - into single -

518 """

519 value = (

520 unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")

521 )

522

523 # Return username if it matches our standards

524 if USERNAME_MATCHER.match(value):

525 return value

526

527 value = STRIP_MATCHER.sub("", value).strip().lstrip(".")

528 return CLEANUP_MATCHER.sub("-", value)

529

530

531def cycle_session(strategy, user: User, *args, **kwargs) -> None:

532 # Change key for current session and invalidate others

533 cycle_session_keys(strategy.request, user)

534

535

536def adjust_primary_mail(strategy, entries, user: User, *args, **kwargs) -> None:

537 """Fix primary mail on disconnect."""

538 # Remove pending verification codes

539 invalidate_reset_codes(user=user, entries=entries)

540

541 # Check remaining verified mails

542 verified = VerifiedEmail.objects.filter(social__user=user).exclude(

543 social__in=entries

544 )

545 if verified.filter(email=user.email).exists():

546 return

547

548 user.email = verified[0].email

549 user.save()

550 messages.warning(

551 strategy.request,

552 gettext(

553 "Your e-mail no longer belongs to verified account, "

554 "it has been changed to {0}."

555 ).format(user.email),

556 )

557

558

559def notify_disconnect(strategy, backend, entries, user: User, **kwargs) -> None:

560 """Store verified e-mail."""

561 for social in entries:

562 AuditLog.objects.create(

563 user,

564 strategy.request,

565 "auth-disconnect",

566 method=backend.name,

567 name=social.uid,

568 )

569

570

571@partial

572def second_factor(strategy, backend, user: User, current_partial, **kwargs):

573 """Force authentication when adding new association."""

574 if user.profile.has_2fa and DEVICE_ID_SESSION_KEY not in strategy.request.session:

575 # Store session indication for second factor

576 strategy.request.session[SESSION_SECOND_FACTOR_USER] = (user.id, "")

577 strategy.request.session[SESSION_SECOND_FACTOR_SOCIAL] = True

578 # Redirect to second factor login

579 continue_url = "{}?partial_token={}".format(

580 reverse("social:complete", args=(backend.name,)), current_partial.token

581 )

582 login_params = {"next": continue_url}

583 login_url = reverse(

584 "2fa-login", kwargs={"backend": user.profile.get_second_factor_type()}

585 )

586 return HttpResponseRedirect(f"{login_url}?{urlencode(login_params)}")

587 return None