defmodule GitGud.Accounts.TwoFactor do
@moduledoc """
Time-based one-time password (RFC 6238) + recovery codes for users.
Three states a user can be in:
1. **Disabled** — no `totp_*` fields set. Login skips the second
factor step entirely.
2. **Pending** — `totp_ciphertext` is set but `totp_enabled_at`
is nil. Happens during enrollment: a fresh secret is stored
but the user hasn't proven they can produce a valid code yet.
Login still skips the second factor — a half-enrolled secret
must never lock a user out.
3. **Enabled** — both set. Login requires a valid 6-digit TOTP code
or a one-shot recovery code.
Secrets at rest go through `GitGud.Secrets.encrypt/1` (AES-256-GCM with
the configured master key + per-row IV), the same primitive used for
CI secrets.
Recovery codes are 10 random 10-character strings hashed with bcrypt.
Verification looks them up by the bcrypt-friendly comparison rather
than equality (each row is its own salt). A code's row is marked
`used_at` on first valid use; re-presentation fails.
"""
import Ecto.Query, warn: false
alias GitGud.Accounts.RecoveryCode
alias GitGud.Accounts.User
alias GitGud.Repo
alias GitGud.Secrets
@recovery_code_count 10
@recovery_code_bytes 6
# ── enrollment ──────────────────────────────────────────────────────
@doc """
Generate a fresh TOTP secret. Returned in raw bytes; the caller is
expected to keep it in session (or a stash) until the user confirms
enrollment with a valid code via `enable/3`.
"""
def new_secret, do: NimbleTOTP.secret()
@doc """
Build the `otpauth://` URI used to seed authenticator apps. Usually
rendered as a QR code via `qr_svg/1`.
"""
def provisioning_uri(secret, %User{} = user) do
NimbleTOTP.otpauth_uri("GitGud:#{user.email}", secret, issuer: "GitGud")
end
@doc "Render a `provisioning_uri/2` value as an inline SVG QR code."
def qr_svg(uri) do
uri
|> EQRCode.encode()
|> EQRCode.svg(width: 220, background_color: "#ffffff", color: "#000000")
end
@doc """
Verify `code` against `secret` and, on match, persist the encrypted
secret + mint recovery codes. Returns `{:ok, plaintext_codes}` (the
one and only time the user sees the codes) or `{:error, :invalid_code}`.
"""
def enable(%User{} = user, secret, code) when is_binary(secret) and is_binary(code) do
if NimbleTOTP.valid?(secret, code) do
do_enable(user, secret)
else
{:error, :invalid_code}
end
end
defp do_enable(user, secret) do
{:ok, enc} = Secrets.encrypt(secret)
now = DateTime.utc_now(:second)
Repo.transaction(fn ->
updated =
user
|> Ecto.Changeset.change(
totp_key_id: enc.key_id,
totp_ciphertext: enc.ciphertext,
totp_iv: enc.iv,
totp_tag: enc.tag,
totp_enabled_at: now
)
|> Repo.update!()
codes = regenerate_recovery_codes!(updated)
{updated, codes}
end)
|> case do
{:ok, {_user, codes}} -> {:ok, codes}
err -> err
end
end
@doc """
Turn off TOTP for `user`. Clears the secret. Recovery codes are
*kept* if the user still has a WebAuthn credential, since they
remain a valid fallback for the remaining factor; otherwise they're
invalidated since 2FA is fully off.
"""
def disable(%User{} = user) do
Repo.transaction(fn ->
unless GitGud.Accounts.Webauthn.has_credentials?(user) do
Repo.delete_all(from r in RecoveryCode, where: r.user_id == ^user.id)
end
user
|> Ecto.Changeset.change(
totp_key_id: nil,
totp_ciphertext: nil,
totp_iv: nil,
totp_tag: nil,
totp_enabled_at: nil
)
|> Repo.update!()
end)
|> case do
{:ok, updated} -> {:ok, updated}
err -> err
end
end
# ── verification ────────────────────────────────────────────────────
@doc "True if the user has finished enrolling TOTP."
def enabled?(%User{} = user), do: User.totp_enabled?(user)
@doc """
Verify a 6-digit TOTP code against the user's stored secret. Returns
`:ok` or `{:error, :invalid_code}` / `{:error, :not_enrolled}`.
NimbleTOTP tolerates a single ±30s window of clock drift by default.
"""
def verify_code(%User{totp_ciphertext: nil}, _code), do: {:error, :not_enrolled}
def verify_code(%User{} = user, code) when is_binary(code) do
with {:ok, secret} <- decrypt_secret(user) do
if NimbleTOTP.valid?(secret, code) do
:ok
else
{:error, :invalid_code}
end
end
end
defp decrypt_secret(user) do
Secrets.decrypt(%{
ciphertext: user.totp_ciphertext,
iv: user.totp_iv,
tag: user.totp_tag,
key_id: user.totp_key_id
})
end
# ── recovery codes ──────────────────────────────────────────────────
@doc """
Ensure the user has at least one unused recovery code; if not, mint
a fresh batch and return the plaintext codes (the only time the user
sees them). If they already have codes, returns `nil`.
Called from the first-enrollment paths of any factor type (TOTP
confirm + WebAuthn first credential) so the codes are issued
exactly once across MFA setup methods.
"""
def ensure_recovery_codes(%User{} = user) do
if remaining_recovery_codes(user) > 0 do
nil
else
regenerate_recovery_codes!(user)
end
end
@doc """
Replace all of `user`'s recovery codes with a fresh batch. Returns the
plaintext codes — the only chance the user gets to record them.
"""
def regenerate_recovery_codes!(%User{id: uid} = _user) do
Repo.delete_all(from r in RecoveryCode, where: r.user_id == ^uid)
plaintexts = for _ <- 1..@recovery_code_count, do: generate_code()
rows =
Enum.map(plaintexts, fn code ->
%{
user_id: uid,
code_hash: Bcrypt.hash_pwd_salt(code),
inserted_at: DateTime.utc_now(:second)
}
end)
{_, _} = Repo.insert_all(RecoveryCode, rows)
plaintexts
end
defp generate_code do
:crypto.strong_rand_bytes(@recovery_code_bytes)
|> Base.encode32(case: :lower, padding: false)
|> String.slice(0, 10)
end
@doc """
Consume a recovery code. Verifies the supplied `code` against any of
the user's unused codes; on a hit marks that row used and returns
`:ok`. No-hit + already-used both return `{:error, :invalid_code}`.
"""
def verify_recovery_code(%User{id: uid}, code) when is_binary(code) do
rows =
Repo.all(
from r in RecoveryCode,
where: r.user_id == ^uid and is_nil(r.used_at)
)
case Enum.find(rows, fn r -> Bcrypt.verify_pass(code, r.code_hash) end) do
nil ->
# Run the bcrypt cost anyway to avoid leaking validity via timing.
_ = Bcrypt.no_user_verify()
{:error, :invalid_code}
row ->
{:ok, _} = row |> RecoveryCode.mark_used() |> Repo.update()
:ok
end
end
@doc "How many unused recovery codes does the user have?"
def remaining_recovery_codes(%User{id: uid}) do
Repo.aggregate(
from(r in RecoveryCode, where: r.user_id == ^uid and is_nil(r.used_at)),
:count
)
end
# ── instance-level enforcement ──────────────────────────────────────
@doc """
Is TOTP required by instance policy for any new account? Set via:
config :git_gud, GitGud.Accounts.TwoFactor,
required: true,
grace_period_days: 14
"""
def required? do
cfg() |> Keyword.get(:required, false)
end
@doc """
Days from `user.confirmed_at` (or `inserted_at` as fallback) during
which the user can still use the forge without enrolling. After this
period, on_mount hooks redirect them to the enrollment page until
they complete it. Default is 14 days.
"""
def grace_period_days do
cfg() |> Keyword.get(:grace_period_days, 14)
end
@doc """
Status check for an authed user.
* `:not_required` — instance doesn't require 2FA
* `:enrolled` — user already has TOTP on
* `{:grace, days_left}` — required, not enrolled, still inside grace
* `:enforced` — required, not enrolled, grace expired
"""
def enforcement_status(%User{} = user) do
cond do
not required?() -> :not_required
enabled?(user) -> :enrolled
true -> in_grace?(user)
end
end
defp in_grace?(%User{} = user) do
anchor = user.confirmed_at || user.inserted_at
case anchor do
%DateTime{} = dt ->
elapsed_seconds = DateTime.diff(DateTime.utc_now(), dt)
grace_seconds = grace_period_days() * 86_400
if elapsed_seconds >= grace_seconds do
:enforced
else
{:grace, max(0, div(grace_seconds - elapsed_seconds, 86_400))}
end
_ ->
# No reliable anchor — apply grace freshly from now.
{:grace, grace_period_days()}
end
end
defp cfg do
Application.get_env(:git_gud, __MODULE__, [])
end
end
9.2 KiB · text
5af55d9