<s> """<STR_LIT>""" <EOL> import gzip <EOL> import hmac <EOL> import json <EOL> import logging <EOL> import ssl <EOL> import threading <EOL> import time <EOL> from datetime import timedelta <EOL> from http import cookies <EOL> from http . server import HTTPServer , SimpleHTTPRequestHandler <EOL> from socketserver import ThreadingMixIn <EOL> from urllib . parse import parse_qs , urlparse <EOL> import homeassistant . bootstrap as bootstrap <EOL> import homeassistant . core as ha <EOL> import homeassistant . remote as rem <EOL> import homeassistant . util as util <EOL> import homeassistant . util . dt as date_util <EOL> from homeassistant . const import ( <EOL> CONTENT_TYPE_JSON , CONTENT_TYPE_TEXT_PLAIN , HTTP_HEADER_ACCEPT_ENCODING , <EOL> HTTP_HEADER_CACHE_CONTROL , HTTP_HEADER_CONTENT_ENCODING , <EOL> HTTP_HEADER_CONTENT_LENGTH , HTTP_HEADER_CONTENT_TYPE , HTTP_HEADER_EXPIRES , <EOL> HTTP_HEADER_HA_AUTH , HTTP_HEADER_VARY , HTTP_METHOD_NOT_ALLOWED , <EOL> HTTP_NOT_FOUND , HTTP_OK , HTTP_UNAUTHORIZED , HTTP_UNPROCESSABLE_ENTITY , <EOL> SERVER_PORT , URL_ROOT , URL_API_EVENT_FORWARD ) <EOL> DOMAIN = "<STR_LIT:http>" <EOL> CONF_API_PASSWORD = "<STR_LIT>" <EOL> CONF_SERVER_HOST = "<STR_LIT>" <EOL> CONF_SERVER_PORT = "<STR_LIT>" <EOL> CONF_DEVELOPMENT = "<STR_LIT>" <EOL> CONF_SSL_CERTIFICATE = '<STR_LIT>' <EOL> CONF_SSL_KEY = '<STR_LIT>' <EOL> DATA_API_PASSWORD = '<STR_LIT>' <EOL> SESSION_CLEAR_INTERVAL = timedelta ( seconds = <NUM_LIT:20> ) <EOL> SESSION_TIMEOUT_SECONDS = <NUM_LIT> <EOL> SESSION_KEY = '<STR_LIT>' <EOL> _LOGGER = logging . getLogger ( __name__ ) <EOL> def setup ( hass , config ) : <EOL> """<STR_LIT>""" <EOL> conf = config . get ( DOMAIN , { } ) <EOL> api_password = util . convert ( conf . get ( CONF_API_PASSWORD ) , str ) <EOL> server_host = conf . get ( CONF_SERVER_HOST , '<STR_LIT>' ) <EOL> server_port = conf . get ( CONF_SERVER_PORT , SERVER_PORT ) <EOL> development = str ( conf . get ( CONF_DEVELOPMENT , "<STR_LIT>" ) ) == "<STR_LIT:1>" <EOL> ssl_certificate = conf . get ( CONF_SSL_CERTIFICATE ) <EOL> ssl_key = conf . get ( CONF_SSL_KEY ) <EOL> try : <EOL> server = HomeAssistantHTTPServer ( <EOL> ( server_host , server_port ) , RequestHandler , hass , api_password , <EOL> development , ssl_certificate , ssl_key ) <EOL> except OSError : <EOL> _LOGGER . exception ( "<STR_LIT>" ) <EOL> return False <EOL> hass . bus . listen_once ( <EOL> ha . EVENT_HOMEASSISTANT_START , <EOL> lambda event : <EOL> threading . Thread ( target = server . start , daemon = True , <EOL> name = '<STR_LIT>' ) . start ( ) ) <EOL> hass . http = server <EOL> hass . config . api = rem . API ( server_host if server_host != '<STR_LIT>' <EOL> else util . get_local_ip ( ) , <EOL> api_password , server_port , <EOL> ssl_certificate is not None ) <EOL> return True <EOL> class HomeAssistantHTTPServer ( ThreadingMixIn , HTTPServer ) : <EOL> """<STR_LIT>""" <EOL> allow_reuse_address = True <EOL> daemon_threads = True <EOL> def __init__ ( self , server_address , request_handler_class , <EOL> hass , api_password , development , ssl_certificate , ssl_key ) : <EOL> """<STR_LIT>""" <EOL> super ( ) . __init__ ( server_address , request_handler_class ) <EOL> self . server_address = server_address <EOL> self . hass = hass <EOL> self . api_password = api_password <EOL> self . development = development <EOL> self . paths = [ ] <EOL> self . sessions = SessionStore ( ) <EOL> self . use_ssl = ssl_certificate is not None <EOL> self . event_forwarder = None <EOL> if development : <EOL> _LOGGER . info ( "<STR_LIT>" ) <EOL> if ssl_certificate is not None : <EOL> context = ssl . create_default_context ( <EOL> purpose = ssl . Purpose . CLIENT_AUTH ) <EOL> context . load_cert_chain ( ssl_certificate , keyfile = ssl_key ) <EOL> self . socket = context . wrap_socket ( self . socket , server_side = True ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> def stop_http ( event ) : <EOL> """<STR_LIT>""" <EOL> self . shutdown ( ) <EOL> self . hass . bus . listen_once ( ha . EVENT_HOMEASSISTANT_STOP , stop_http ) <EOL> protocol = '<STR_LIT>' if self . use_ssl else '<STR_LIT:http>' <EOL> _LOGGER . info ( <EOL> "<STR_LIT>" , <EOL> protocol , self . server_address [ <NUM_LIT:0> ] , self . server_address [ <NUM_LIT:1> ] ) <EOL> bootstrap . setup_component ( self . hass , '<STR_LIT>' ) <EOL> bootstrap . setup_component ( self . hass , '<STR_LIT>' ) <EOL> self . serve_forever ( ) <EOL> def register_path ( self , method , url , callback , require_auth = True ) : <EOL> """<STR_LIT>""" <EOL> self . paths . append ( ( method , url , callback , require_auth ) ) <EOL> def log_message ( self , fmt , * args ) : <EOL> """<STR_LIT>""" <EOL> _LOGGER . info ( fmt , * args ) <EOL> class RequestHandler ( SimpleHTTPRequestHandler ) : <EOL> """<STR_LIT>""" <EOL> server_version = "<STR_LIT>" <EOL> def __init__ ( self , req , client_addr , server ) : <EOL> """<STR_LIT>""" <EOL> self . authenticated = False <EOL> SimpleHTTPRequestHandler . __init__ ( self , req , client_addr , server ) <EOL> self . protocol_version = '<STR_LIT>' <EOL> def log_message ( self , fmt , * arguments ) : <EOL> """<STR_LIT>""" <EOL> if self . server . api_password is None : <EOL> _LOGGER . info ( fmt , * arguments ) <EOL> else : <EOL> _LOGGER . info ( <EOL> fmt , * ( arg . replace ( self . server . api_password , '<STR_LIT>' ) <EOL> if isinstance ( arg , str ) else arg for arg in arguments ) ) <EOL> def _handle_request ( self , method ) : <EOL> """<STR_LIT>""" <EOL> url = urlparse ( self . path ) <EOL> data = { key : data [ - <NUM_LIT:1> ] for key , data in parse_qs ( url . query ) . items ( ) } <EOL> content_length = int ( self . headers . get ( HTTP_HEADER_CONTENT_LENGTH , <NUM_LIT:0> ) ) <EOL> if content_length : <EOL> body_content = self . rfile . read ( content_length ) . decode ( "<STR_LIT>" ) <EOL> try : <EOL> data . update ( json . loads ( body_content ) ) <EOL> except ( TypeError , ValueError ) : <EOL> _LOGGER . exception ( <EOL> "<STR_LIT>" , body_content ) <EOL> self . write_json_message ( <EOL> "<STR_LIT>" , HTTP_UNPROCESSABLE_ENTITY ) <EOL> return <EOL> if self . verify_session ( ) : <EOL> self . authenticated = True <EOL> elif self . server . api_password is None : <EOL> self . authenticated = True <EOL> elif hmac . compare_digest ( self . headers . get ( HTTP_HEADER_HA_AUTH , '<STR_LIT>' ) , <EOL> self . server . api_password ) : <EOL> self . authenticated = True <EOL> elif hmac . compare_digest ( data . get ( DATA_API_PASSWORD , '<STR_LIT>' ) , <EOL> self . server . api_password ) : <EOL> self . authenticated = True <EOL> else : <EOL> self . authenticated = False <EOL> if url . path not in [ URL_ROOT , URL_API_EVENT_FORWARD ] : <EOL> data . pop ( DATA_API_PASSWORD , None ) <EOL> if '<STR_LIT>' in data : <EOL> method = data . pop ( '<STR_LIT>' ) <EOL> path_matched_but_not_method = False <EOL> handle_request_method = False <EOL> require_auth = True <EOL> for t_method , t_path , t_handler , t_auth in self . server . paths : <EOL> if isinstance ( t_path , str ) : <EOL> path_match = url . path == t_path <EOL> else : <EOL> path_match = t_path . match ( url . path ) <EOL> if path_match and method == t_method : <EOL> handle_request_method = t_handler <EOL> require_auth = t_auth <EOL> break <EOL> elif path_match : <EOL> path_matched_but_not_method = True <EOL> if handle_request_method : <EOL> msg = "<STR_LIT>" <EOL> if require_auth and not self . authenticated : <EOL> self . write_json_message ( msg , HTTP_UNAUTHORIZED ) <EOL> _LOGGER . warning ( '<STR_LIT>' , <EOL> msg , <EOL> self . client_address [ <NUM_LIT:0> ] ) <EOL> return <EOL> handle_request_method ( self , path_match , data ) <EOL> elif path_matched_but_not_method : <EOL> self . send_response ( HTTP_METHOD_NOT_ALLOWED ) <EOL> self . end_headers ( ) <EOL> else : <EOL> self . send_response ( HTTP_NOT_FOUND ) <EOL> self . end_headers ( ) <EOL> def do_HEAD ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _handle_request ( '<STR_LIT>' ) <EOL> def do_GET ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _handle_request ( '<STR_LIT:GET>' ) <EOL> def do_POST ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _handle_request ( '<STR_LIT:POST>' ) <EOL> def do_PUT ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _handle_request ( '<STR_LIT>' ) <EOL> def do_DELETE ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _handle_request ( '<STR_LIT>' ) <EOL> def write_json_message ( self , message , status_code = HTTP_OK ) : <EOL> """<STR_LIT>""" <EOL> self . write_json ( { '<STR_LIT:message>' : message } , status_code = status_code ) <EOL> def write_json ( self , data = None , status_code = HTTP_OK , location = None ) : <EOL> """<STR_LIT>""" <EOL> json_data = json . dumps ( data , indent = <NUM_LIT:4> , sort_keys = True , <EOL> cls = rem . JSONEncoder ) . encode ( '<STR_LIT>' ) <EOL> self . send_response ( status_code ) <EOL> if location : <EOL> self . send_header ( '<STR_LIT>' , location ) <EOL> self . set_session_cookie_header ( ) <EOL> self . write_content ( json_data , CONTENT_TYPE_JSON ) <EOL> def write_text ( self , message , status_code = HTTP_OK ) : <EOL> """<STR_LIT>""" <EOL> msg_data = message . encode ( '<STR_LIT>' ) <EOL> self . send_response ( status_code ) <EOL> self . set_session_cookie_header ( ) <EOL> self . write_content ( msg_data , CONTENT_TYPE_TEXT_PLAIN ) <EOL> def write_file ( self , path , cache_headers = True ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> with open ( path , '<STR_LIT:rb>' ) as inp : <EOL> self . write_file_pointer ( self . guess_type ( path ) , inp , <EOL> cache_headers ) <EOL> except IOError : <EOL> self . send_response ( HTTP_NOT_FOUND ) <EOL> self . end_headers ( ) <EOL> _LOGGER . exception ( "<STR_LIT>" , path ) <EOL> def write_file_pointer ( self , content_type , inp , cache_headers = True ) : <EOL> """<STR_LIT>""" <EOL> self . send_response ( HTTP_OK ) <EOL> if cache_headers : <EOL> self . set_cache_header ( ) <EOL> self . set_session_cookie_header ( ) <EOL> self . write_content ( inp . read ( ) , content_type ) <EOL> def write_content ( self , content , content_type = None ) : <EOL> """<STR_LIT>""" <EOL> if content_type is not None : <EOL> self . send_header ( HTTP_HEADER_CONTENT_TYPE , content_type ) <EOL> if '<STR_LIT>' in self . headers . get ( HTTP_HEADER_ACCEPT_ENCODING , '<STR_LIT>' ) : <EOL> content = gzip . compress ( content ) <EOL> self . send_header ( HTTP_HEADER_CONTENT_ENCODING , "<STR_LIT>" ) <EOL> self . send_header ( HTTP_HEADER_VARY , HTTP_HEADER_ACCEPT_ENCODING ) <EOL> self . send_header ( HTTP_HEADER_CONTENT_LENGTH , str ( len ( content ) ) ) <EOL> self . end_headers ( ) <EOL> if self . command == '<STR_LIT>' : <EOL> return <EOL> self . wfile . write ( content ) <EOL> def set_cache_header ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . server . development : <EOL> return <EOL> cache_time = <NUM_LIT> * <NUM_LIT> <EOL> self . send_header ( <EOL> HTTP_HEADER_CACHE_CONTROL , <EOL> "<STR_LIT>" . format ( cache_time ) ) <EOL> self . send_header ( <EOL> HTTP_HEADER_EXPIRES , <EOL> self . date_time_string ( time . time ( ) + cache_time ) ) <EOL> def set_session_cookie_header ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . authenticated : <EOL> return None <EOL> session_id = self . get_cookie_session_id ( ) <EOL> if session_id is not None : <EOL> self . server . sessions . extend_validation ( session_id ) <EOL> return session_id <EOL> self . send_header ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' . format ( SESSION_KEY , self . server . sessions . create ( ) ) <EOL> ) <EOL> return session_id <EOL> def verify_session ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . get_cookie_session_id ( ) is not None <EOL> def get_cookie_session_id ( self ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' not in self . headers : <EOL> return None <EOL> cookie = cookies . SimpleCookie ( ) <EOL> try : <EOL> cookie . load ( self . headers [ "<STR_LIT>" ] ) <EOL> except cookies . CookieError : <EOL> return None <EOL> morsel = cookie . get ( SESSION_KEY ) <EOL> if morsel is None : <EOL> return None <EOL> session_id = cookie [ SESSION_KEY ] . value <EOL> if self . server . sessions . is_valid ( session_id ) : <EOL> return session_id <EOL> return None <EOL> def destroy_session ( self ) : <EOL> """<STR_LIT>""" <EOL> session_id = self . get_cookie_session_id ( ) <EOL> if session_id is None : <EOL> return <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . server . sessions . destroy ( session_id ) <EOL> def session_valid_time ( ) : <EOL> """<STR_LIT>""" <EOL> return date_util . utcnow ( ) + timedelta ( seconds = SESSION_TIMEOUT_SECONDS ) <EOL> class SessionStore ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _sessions = { } <EOL> self . _lock = threading . RLock ( ) <EOL> @ util . Throttle ( SESSION_CLEAR_INTERVAL ) <EOL> def _remove_expired ( self ) : <EOL> """<STR_LIT>""" <EOL> now = date_util . utcnow ( ) <EOL> for key in [ key for key , valid_time in self . _sessions . items ( ) <EOL> if valid_time < now ] : <EOL> self . _sessions . pop ( key ) <EOL> def is_valid ( self , key ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> self . _remove_expired ( ) <EOL> return ( key in self . _sessions and <EOL> self . _sessions [ key ] > date_util . utcnow ( ) ) <EOL> def extend_validation ( self , key ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> if key not in self . _sessions : <EOL> return <EOL> self . _sessions [ key ] = session_valid_time ( ) <EOL> def destroy ( self , key ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> self . _sessions . pop ( key , None ) <EOL> def create ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> session_id = util . get_random_string ( <NUM_LIT:20> ) <EOL> while session_id in self . _sessions : <EOL> session_id = util . get_random_string ( <NUM_LIT:20> ) <EOL> self . _sessions [ session_id ] = session_valid_time ( ) <EOL> return session_id </s>
