<s> from u2fval . model import Client , User , Device <EOL> from u2fval . core . exc import ( BadInputException , NoEligibleDevicesException , <EOL> DeviceCompromisedException ) <EOL> from u2flib_server . u2f_v2 import ( start_register , complete_register , <EOL> start_authenticate , verify_authenticate ) <EOL> from u2flib_server . utils import rand_bytes <EOL> from sqlalchemy . orm import exc <EOL> from datetime import datetime <EOL> import logging <EOL> __all__ = [ '<STR_LIT>' ] <EOL> log = logging . getLogger ( __name__ ) <EOL> class U2FController ( object ) : <EOL> def __init__ ( self , session , memstore , client_name , metadata , <EOL> require_trusted = True ) : <EOL> self . _session = session <EOL> self . _memstore = memstore <EOL> try : <EOL> self . _client = session . query ( Client ) . filter ( Client . name == client_name ) . one ( ) <EOL> except exc . NoResultFound : <EOL> raise KeyError ( '<STR_LIT>' % client_name ) <EOL> self . _metadata = metadata <EOL> self . _require_trusted = require_trusted <EOL> def _get_user ( self , username ) : <EOL> return self . _session . query ( User ) . filter ( User . client_id == self . _client . id ) . filter ( User . name == username ) . first ( ) <EOL> def _get_device ( self , handle ) : <EOL> return self . _session . query ( Device ) . join ( Device . user ) . filter ( User . client_id == self . _client . id ) . filter ( Device . handle == handle ) . first ( ) <EOL> def _get_or_create_user ( self , username ) : <EOL> user = self . _get_user ( username ) <EOL> if user is None : <EOL> user = User ( username ) <EOL> self . _client . users . append ( user ) <EOL> log . info ( '<STR_LIT>' , self . _client . name , username ) <EOL> return user <EOL> @ property <EOL> def client_name ( self ) : <EOL> return self . _client . name <EOL> def get_trusted_facets ( self ) : <EOL> return { <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT:version>' : { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:0> } , <EOL> '<STR_LIT>' : self . _client . valid_facets <EOL> } ] <EOL> } <EOL> def delete_user ( self , username ) : <EOL> user = self . _get_user ( username ) <EOL> if user is not None : <EOL> self . _session . delete ( user ) <EOL> log . info ( '<STR_LIT>' , self . _client . name , username ) <EOL> def register_start ( self , username ) : <EOL> register_request = start_register ( self . _client . app_id ) <EOL> self . _memstore . store ( self . _client . id , username , <EOL> register_request . challenge , <EOL> { '<STR_LIT>' : register_request } ) <EOL> sign_requests = [ ] <EOL> descriptors = [ ] <EOL> user = self . _get_user ( username ) <EOL> if user is not None : <EOL> for dev in user . devices . values ( ) : <EOL> sign_requests . append ( <EOL> start_authenticate ( dev . bind_data , '<STR_LIT>' ) ) <EOL> descriptors . append ( dev . get_descriptor ( <EOL> self . _metadata . get_metadata ( dev ) ) ) <EOL> return [ register_request ] , sign_requests , descriptors <EOL> def register_complete ( self , username , resp ) : <EOL> memkey = resp . clientData . challenge <EOL> data = self . _memstore . retrieve ( self . _client . id , username , memkey ) <EOL> bind , cert = complete_register ( data [ '<STR_LIT>' ] , resp , <EOL> self . _client . valid_facets ) <EOL> attestation = self . _metadata . get_attestation ( cert ) <EOL> if self . _require_trusted and not attestation . trusted : <EOL> raise BadInputException ( '<STR_LIT>' ) <EOL> user = self . _get_or_create_user ( username ) <EOL> dev = user . add_device ( bind . json , cert , attestation . transports ) <EOL> log . info ( '<STR_LIT>' , <EOL> self . _client . name , username , dev . handle ) <EOL> return dev . handle <EOL> def unregister ( self , handle ) : <EOL> dev = self . _get_device ( handle ) <EOL> self . _session . delete ( dev ) <EOL> log . info ( '<STR_LIT>' , <EOL> self . _client . name , dev . user . name , handle ) <EOL> def set_props ( self , handle , props ) : <EOL> dev = self . _get_device ( handle ) <EOL> dev . properties . update ( props ) <EOL> def _do_get_descriptor ( self , user , handle ) : <EOL> if user is not None : <EOL> dev = self . _session . query ( Device ) . filter ( Device . user_id == user . id ) . filter ( Device . handle == handle ) . first ( ) <EOL> if user is None or dev is None : <EOL> raise BadInputException ( '<STR_LIT>' % handle ) <EOL> return dev . get_descriptor ( self . _metadata . get_metadata ( dev ) ) <EOL> def get_descriptor ( self , username , handle ) : <EOL> user = self . _get_user ( username ) <EOL> return self . _do_get_descriptor ( user , handle ) <EOL> def get_descriptors ( self , username ) : <EOL> user = self . _get_user ( username ) <EOL> if user is None : <EOL> return [ ] <EOL> return [ d . get_descriptor ( self . _metadata . get_metadata ( d ) ) <EOL> for d in user . devices . values ( ) ] <EOL> def authenticate_start ( self , username , invalidate = False ) : <EOL> user = self . _get_user ( username ) <EOL> if user is None or len ( user . devices ) == <NUM_LIT:0> : <EOL> log . info ( '<STR_LIT>' , username ) <EOL> raise NoEligibleDevicesException ( '<STR_LIT>' , [ ] ) <EOL> sign_requests = [ ] <EOL> descriptors = [ ] <EOL> challenges = { } <EOL> rand = rand_bytes ( <NUM_LIT:32> ) <EOL> for handle , dev in user . devices . items ( ) : <EOL> if not dev . compromised : <EOL> challenge = start_authenticate ( dev . bind_data , rand ) <EOL> sign_requests . append ( challenge ) <EOL> descriptors . append ( dev . get_descriptor ( <EOL> self . _metadata . get_metadata ( dev ) ) ) <EOL> challenges [ handle ] = { <EOL> '<STR_LIT>' : challenge . keyHandle , <EOL> '<STR_LIT>' : challenge <EOL> } <EOL> if not sign_requests : <EOL> raise NoEligibleDevicesException ( <EOL> '<STR_LIT>' , <EOL> [ d . get_descriptor ( ) for d in user . devices . values ( ) ] <EOL> ) <EOL> self . _memstore . store ( self . _client . id , username , rand , challenges ) <EOL> return sign_requests , descriptors <EOL> def authenticate_complete ( self , username , resp ) : <EOL> memkey = resp . clientData . challenge <EOL> challenges = self . _memstore . retrieve ( self . _client . id , username , memkey ) <EOL> user = self . _get_user ( username ) <EOL> for handle , data in challenges . items ( ) : <EOL> if data [ '<STR_LIT>' ] == resp . keyHandle : <EOL> dev = user . devices [ handle ] <EOL> if dev . compromised : <EOL> raise BadInputException ( '<STR_LIT>' ) <EOL> counter , presence = verify_authenticate ( <EOL> dev . bind_data , <EOL> data [ '<STR_LIT>' ] , <EOL> resp , <EOL> self . _client . valid_facets <EOL> ) <EOL> if presence == chr ( <NUM_LIT:0> ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> if counter > ( dev . counter or - <NUM_LIT:1> ) : <EOL> dev . counter = counter <EOL> dev . authenticated_at = datetime . now ( ) <EOL> return handle <EOL> dev . compromised = True <EOL> raise DeviceCompromisedException ( '<STR_LIT>' , <EOL> dev . get_descriptor ( ) ) <EOL> else : <EOL> raise BadInputException ( '<STR_LIT>' % <EOL> resp . keyHandle ) </s>
