import base64 import secrets from urllib.parse import urlsplit from datetime import timedelta import base58 from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from django.contrib.auth.hashers import make_password, check_password from django.db import transaction, IntegrityError from django.utils import timezone from rest_framework.exceptions import AuthenticationFailed, ValidationError from .security_models import LoginIdentity, AuthChallenge from .security_config import value from . import security_challenges as challenges from . import security_sms as sms from .security_mfa import verify_factor from .security_sessions import issue def link(user, provider, subject, label=''): try: with transaction.atomic(): identity,created=LoginIdentity.objects.get_or_create(provider=provider,subject=subject, defaults={'user':user,'label':label[:120]}) if identity.user_id!=user.pk: raise ValidationError('This identity is already linked to another account.') return {'linked':True,'identity_id':str(identity.pk),'provider':provider} except IntegrityError as exc: raise ValidationError('An identity is already linked; unlink the old phone first if necessary.') from exc @transaction.atomic def login_identity(provider, subject, request, otp=''): from .models import User identity=LoginIdentity.objects.select_related('user').filter(provider=provider,subject=subject).first() if identity is None or not identity.user.is_active or not identity.user.is_email_verified: raise AuthenticationFailed('No eligible linked account. Sign in with email and link this identity first.') user=User.objects.select_for_update().get(pk=identity.user_id) verify_factor(user,otp) result=issue(user,request,provider) identity.last_used_at=timezone.now();identity.save(update_fields=['last_used_at']) return result def wallet_begin(origin, address, user=None): try: public=base58.b58decode(address) if len(public)!=32 or base58.b58encode(public).decode()!=address: raise ValueError except (ValueError,TypeError) as exc: raise ValidationError('A canonical Solana public address is required.') from exc now=timezone.now();nonce=secrets.token_hex(24) action='link this wallet to your FTG account' if user else 'sign in to FTG' message=(f'{urlsplit(origin).netloc} wants you to sign in with your Solana account:\n{address}\n\n' f'Sign to {action}. This does not authorize a transaction or transfer.\n\nURI: {origin}\nVersion: 1\n' f'Chain ID: {value("ACCOUNT_SOLANA_CHAIN","solana:mainnet")}\nNonce: {nonce}\nIssued At: {now.isoformat()}\n' f'Expiration Time: {(now+timedelta(minutes=5)).isoformat()}') row=challenges.create('SOLANA_LINK' if user else 'SOLANA_LOGIN',{'message':message,'address':address,'origin':origin},user) return {'challenge_id':str(row.pk),'message':message,'address':address,'network':'SOLANA'} def wallet_finish(challenge_id, signature, origin, request, user=None, otp=''): def operation(row): if row.payload['origin']!=origin: raise ValidationError('Wallet origin mismatch.') try: raw=base64.b64decode(signature,validate=True) Ed25519PublicKey.from_public_bytes(base58.b58decode(row.payload['address'])).verify(raw,row.payload['message'].encode()) except (ValueError,TypeError,InvalidSignature) as exc: raise ValidationError('Invalid Solana signature.') from exc if user: return link(user,'SOLANA',row.payload['address'],'Solana wallet') return login_identity('SOLANA',row.payload['address'],request,otp) return challenges.consume(challenge_id,'SOLANA_LINK' if user else 'SOLANA_LOGIN',operation,user) def google_begin(origin, user=None): if not value('ACCOUNT_GOOGLE_CLIENT_ID'): raise ValidationError('Google sign-in is not configured.') nonce=secrets.token_urlsafe(32) row=challenges.create('GOOGLE_LINK' if user else 'GOOGLE_LOGIN',{'nonce':nonce,'origin':origin},user) return {'challenge_id':str(row.pk),'nonce':nonce,'client_id':value('ACCOUNT_GOOGLE_CLIENT_ID')} def google_claims(credential): from google.oauth2 import id_token from google.auth.transport.requests import Request from google.auth.exceptions import GoogleAuthError import requests try: class BoundedRequest(Request): def __call__(self,*args,**kwargs): kwargs['timeout']=8 return super().__call__(*args,**kwargs) if not value('ACCOUNT_GOOGLE_CLIENT_ID'): raise ValueError('Google sign-in is not configured.') claims=id_token.verify_oauth2_token(credential,BoundedRequest(),audience=value('ACCOUNT_GOOGLE_CLIENT_ID')) except (ValueError,GoogleAuthError,requests.RequestException) as exc: raise ValidationError('Google credential could not be verified.') from exc if claims.get('email_verified') is not True or not claims.get('sub'): raise ValidationError('Google email is not verified.') return claims def google_finish(challenge_id,credential,origin,request,user=None,otp=''): # Public-key retrieval takes place before locking the challenge row. claims=google_claims(credential) def operation(row): if row.payload['origin']!=origin or not secrets.compare_digest(str(claims.get('nonce','')),row.payload['nonce']): raise ValidationError('Google nonce or origin mismatch.') if user: return link(user,'GOOGLE',claims['sub'],claims.get('email','')) return login_identity('GOOGLE',claims['sub'],request,otp) return challenges.consume(challenge_id,'GOOGLE_LINK' if user else 'GOOGLE_LOGIN',operation,user) def phone_begin(phone,user=None): phone=sms.normalize(phone) if not sms.configured(): raise sms.SMSUnavailable() # Shared across IPs. Never allow one phone to be spammed via new challenges. from django.core.cache import cache from django.utils.crypto import salted_hmac bucket=salted_hmac('account-phone-rate',phone,algorithm='sha256').hexdigest() if not cache.add('phone-otp:'+bucket,'1',timeout=60): raise ValidationError('Wait one minute before requesting another code.') code=f'{secrets.randbelow(1000000):06d}' purpose='PHONE_LINK' if user else 'PHONE_LOGIN' # For unknown phones return an indistinguishable unusable challenge without sending. eligible=user is not None or LoginIdentity.objects.filter(provider='PHONE',subject=phone,user__is_active=True,user__is_email_verified=True).exists() row=challenges.create(purpose,{'phone':phone,'code_hash':make_password(code),'eligible':eligible},user) if eligible: try: sms.send(phone,code) except sms.SMSUnavailable: AuthChallenge.objects.filter(pk=row.pk).update(consumed_at=timezone.now()) raise return {'challenge_id':str(row.pk),'detail':'If eligible, a verification code has been sent.'} def phone_finish(challenge_id,code,request,user=None,otp=''): def operation(row): if not row.payload.get('eligible') or not check_password(code,row.payload['code_hash']): raise ValidationError('Invalid verification code.') if user: return link(user,'PHONE',row.payload['phone'],'Verified phone') return login_identity('PHONE',row.payload['phone'],request,otp) return challenges.consume(challenge_id,'PHONE_LINK' if user else 'PHONE_LOGIN',operation,user)