Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 118 additions & 1 deletion backend/app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,20 @@

from .database import database_readiness, get_session, init_db
from .locales import resolve_locale
from .models import AuthSessionDB, BridgeAuthNonceDB, CalorieAppUserDB, FoodLogDB
from .models import (
AuthSessionDB,
BridgeAuthNonceDB,
CalorieAppUserDB,
ExternalIdentityDB,
FoodLogDB,
OriginLoginHandoffDB,
)
from .schemas import (
AccountDataExportResponse,
AccountExportAccount,
AccountExportAuthSession,
AccountExportExternalIdentity,
AccountExportLoginHandoff,
CurrentUserResponse,
FoodLog,
FoodLogCreate,
Expand Down Expand Up @@ -957,6 +969,111 @@ def identity_me(
)


@app.get("/api/identity/export", response_model=AccountDataExportResponse)
def identity_export(
response: Response,
session: DbSession,
current_user: CurrentUser,
) -> AccountDataExportResponse:
"""Return a versioned private export of linked data without authentication secrets."""
identities = session.exec(
select(ExternalIdentityDB)
.where(ExternalIdentityDB.calorieapp_user_id == current_user.id)
.order_by(ExternalIdentityDB.created_at, ExternalIdentityDB.id)
).all()
external_subjects = [identity.external_subject for identity in identities]

# Authorization activity predates a direct internal-user foreign key. Refuse
# export rather than returning another user's activity when a legacy subject is
# ambiguously linked across providers or accounts.
if external_subjects:
ambiguous_identity = session.exec(
select(ExternalIdentityDB).where(
ExternalIdentityDB.external_subject.in_(external_subjects),
ExternalIdentityDB.calorieapp_user_id != current_user.id,
)
).first()
if ambiguous_identity is not None:
raise HTTPException(
status_code=409,
detail="Account identity requires operator review before export",
)

food_logs = session.exec(
select(FoodLogDB)
.where(FoodLogDB.owner_id == current_user.id)
.order_by(FoodLogDB.created_at, FoodLogDB.id)
).all()
auth_sessions = session.exec(
select(AuthSessionDB)
.where(AuthSessionDB.calorieapp_user_id == current_user.id)
.order_by(AuthSessionDB.created_at, AuthSessionDB.id)
).all()
handoffs = session.exec(
select(OriginLoginHandoffDB)
.where(OriginLoginHandoffDB.calorieapp_user_id == current_user.id)
.order_by(OriginLoginHandoffDB.created_at, OriginLoginHandoffDB.id)
).all()

response.headers["Content-Disposition"] = (
'attachment; filename="calorieapp-account-data-v1.json"'
)

return AccountDataExportResponse(
export_version="calorieapp-account-data-v1",
exported_at=datetime.now(UTC),
account=AccountExportAccount(
user_id=current_user.id,
status=current_user.status,
created_at=current_user.created_at,
updated_at=current_user.updated_at,
),
external_identities=[
AccountExportExternalIdentity(
provider=identity.provider,
external_subject=identity.external_subject,
xrpl_address=identity.xrpl_address,
created_at=identity.created_at,
last_verified_at=identity.last_verified_at,
)
for identity in identities
],
food_logs=[FoodLog.model_validate(entry.model_dump()) for entry in food_logs],
authentication_sessions=[
AccountExportAuthSession(
created_at=auth_session.created_at,
last_seen_at=auth_session.last_seen_at,
expires_at=auth_session.expires_at,
revoked_at=auth_session.revoked_at,
)
for auth_session in auth_sessions
],
# Legacy authorization rows have only a subject, not direct internal-user
# and provider ownership. Preserve the v1 field while withholding rows
# until a migration can prove ownership without inference.
authorization_events=[],
login_handoffs=[
AccountExportLoginHandoff(
status=handoff.status,
created_at=handoff.created_at,
expires_at=handoff.expires_at,
completed_at=handoff.completed_at,
claimed_at=handoff.claimed_at,
failure_code=handoff.failure_code,
)
for handoff in handoffs
],
excluded_security_fields=[
"authorization_code_hash",
"authorization_state",
"login_session_id",
"session_token_hash",
"handoff_state_hash",
"handoff_token_hash",
],
)


@app.post("/api/identity/logout", response_model=LogoutResponse)
def identity_logout(
response: Response,
Expand Down
114 changes: 114 additions & 0 deletions backend/app/schemas.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,120 @@ def serialize_created_at_as_utc(cls, value: datetime) -> datetime:
return _ensure_utc(value)


class AccountExportAccount(BaseModel):
"""Portable non-secret account fields owned by the authenticated user."""

user_id: str
status: str
created_at: datetime
updated_at: datetime

@field_validator("created_at", "updated_at", mode="after")
@classmethod
def normalize_account_timestamps(cls, value: datetime) -> datetime:
return _ensure_utc(value)


class AccountExportExternalIdentity(BaseModel):
"""External identity link included in the user's private export."""

provider: str
external_subject: str
xrpl_address: Optional[str]
created_at: datetime
last_verified_at: datetime

@field_validator("created_at", "last_verified_at", mode="after")
@classmethod
def normalize_identity_timestamps(cls, value: datetime) -> datetime:
return _ensure_utc(value)


class AccountExportAuthSession(BaseModel):
"""Session activity metadata without token hashes or internal identifiers."""

created_at: datetime
last_seen_at: datetime
expires_at: datetime
revoked_at: Optional[datetime]

@field_validator(
"created_at",
"last_seen_at",
"expires_at",
"revoked_at",
mode="after",
)
@classmethod
def normalize_session_timestamps(
cls,
value: Optional[datetime],
) -> Optional[datetime]:
return _ensure_utc(value) if value is not None else None


class AccountExportAuthorizationEvent(BaseModel):
"""Reserved v1 shape for authorization activity with proven ownership."""

external_subject: str
created_at: datetime
expires_at: datetime
used_at: Optional[datetime]
used_by_ip: Optional[str]

@field_validator("created_at", "expires_at", "used_at", mode="after")
@classmethod
def normalize_authorization_timestamps(
cls,
value: Optional[datetime],
) -> Optional[datetime]:
return _ensure_utc(value) if value is not None else None


class AccountExportLoginHandoff(BaseModel):
"""Browser handoff activity without state or handoff-token hashes."""

status: str
created_at: datetime
expires_at: datetime
completed_at: Optional[datetime]
claimed_at: Optional[datetime]
failure_code: Optional[str]

@field_validator(
"created_at",
"expires_at",
"completed_at",
"claimed_at",
mode="after",
)
@classmethod
def normalize_handoff_timestamps(
cls,
value: Optional[datetime],
) -> Optional[datetime]:
return _ensure_utc(value) if value is not None else None


class AccountDataExportResponse(BaseModel):
"""Versioned authenticated CalorieApp account-data export."""

export_version: Literal["calorieapp-account-data-v1"]
exported_at: datetime
account: AccountExportAccount
external_identities: list[AccountExportExternalIdentity]
food_logs: list[FoodLog]
authentication_sessions: list[AccountExportAuthSession]
authorization_events: list[AccountExportAuthorizationEvent]
login_handoffs: list[AccountExportLoginHandoff]
excluded_security_fields: list[str]

@field_validator("exported_at", mode="after")
@classmethod
def normalize_export_timestamp(cls, value: datetime) -> datetime:
return _ensure_utc(value)


class LogoutResponse(BaseModel):
"""Response after logout."""

Expand Down
Loading