(
request: Request,
settings: Annotated[Settings, Depends(get_settings)],
db: AsyncSession = Depends(get_db),
oauth_client=Depends(get_github_oauth_client),
)
| 476 | |
| 477 | @router.get("/github/callback", name="auth_github_callback") |
| 478 | async def auth_github_callback( |
| 479 | request: Request, |
| 480 | settings: Annotated[Settings, Depends(get_settings)], |
| 481 | db: AsyncSession = Depends(get_db), |
| 482 | oauth_client=Depends(get_github_oauth_client), |
| 483 | ): |
| 484 | if not oauth_client.github: |
| 485 | raise HTTPException( |
| 486 | status_code=500, detail="GitHub OAuth client not configured" |
| 487 | ) |
| 488 | |
| 489 | token = await oauth_client.github.authorize_access_token(request) |
| 490 | response = await oauth_client.github.get("user", token=token) |
| 491 | gh_user = response.json() |
| 492 | |
| 493 | user = await get_user_by_provider(db, "github", str(gh_user["id"])) |
| 494 | |
| 495 | if user: |
| 496 | result = await db.execute( |
| 497 | select(UserIdentity).where( |
| 498 | UserIdentity.user_id == user.id, UserIdentity.provider == "github" |
| 499 | ) |
| 500 | ) |
| 501 | github_identity = result.scalar_one_or_none() |
| 502 | if github_identity: |
| 503 | github_identity.access_token = token["access_token"] |
| 504 | github_identity.provider_metadata = { |
| 505 | "login": gh_user["login"], |
| 506 | "name": gh_user.get("name"), |
| 507 | } |
| 508 | else: |
| 509 | email = await get_github_primary_email(oauth_client, token) |
| 510 | if email: |
| 511 | user = await get_user_by_email(db, email) |
| 512 | |
| 513 | if not user: |
| 514 | if email and not await is_email_allowed(email, db): |
| 515 | await notify_denied( |
| 516 | email, |
| 517 | "github", |
| 518 | request, |
| 519 | settings.access_denied_webhook, |
| 520 | ) |
| 521 | flash(request, _(settings.access_denied_message), "error") |
| 522 | return RedirectResponse("/auth/login", status_code=303) |
| 523 | user = await _create_user_with_team( |
| 524 | request, |
| 525 | db, |
| 526 | email=email or f"{gh_user['login']}@github.local", |
| 527 | name=gh_user.get("name"), |
| 528 | username=gh_user["login"], |
| 529 | ) |
| 530 | |
| 531 | github_identity = UserIdentity( |
| 532 | user_id=user.id, |
| 533 | provider="github", |
| 534 | provider_user_id=str(gh_user["id"]), |
| 535 | access_token=token["access_token"], |
nothing calls this directly
no test coverage detected