import http.server import json from mec_database import mec_database # This class will handles any incoming request from the browser class myHandler(http.server.BaseHTTPRequestHandler): __db__ = mec_database() # Handler for the GET requests def do_GET(self): print('>>> do_GET: ', self.path) self.protocol_version = self.request_version # Check HTTP end_headers if self.__check_http_headers__() == False: resp = "{\"problemDetails\": {\t\"type\": \"Bad Request\",\t\"title\": \"N/A\",\t\"status\": 400,\t\"detail\": \"Wrong headers\",\t\"instance\": \"N/A\"}}" self.send_response(400, 'Bad Request') self.send_header('Host', self.headers.get('Host')) self.send_header('Authorization', self.headers.get('Authorization')) self.send_header('Content-Type', 'application/problem+json') l = len(resp) self.send_header('Content-Length', str(l)) self.end_headers() self.wfile.write(bytes(resp, 'utf-8')) return resp, content_type = self.__process__() print('do_GET: resp= ', resp, ', Content_Type= ', content_type) if (resp == None): self.send_response(404, 'Not Found') self.send_header('Host', self.headers.get('Host')) self.send_header('Authorization', self.headers.get('Authorization')) self.send_header('Content-Type', content_type) self.send_header('Content-Length', '0') self.end_headers() else: if resp.find('Not Found') != -1: self.send_response(404, 'Not Found') elif resp.find('Forbidden') != -1: self.send_response(403, 'Forbidden') elif resp.find('Bad Request') != -1: self.send_response(400, 'Bad Request') else: self.send_response(200, 'OK') self.send_header('Host', self.headers.get('Host')) self.send_header('Authorization', self.headers.get('Authorization')) self.send_header('Content-Type', content_type) l = len(resp) self.send_header('Content-Length', str(l)) self.end_headers() self.wfile.write(bytes(resp, 'utf-8')) # End of do_GET # Handler for the POST requests def do_POST(self): print(">>> do_POST: ", self.path) self.protocol_version = self.request_version # Check HTTP end_headers if self.__check_http_headers__() == False: resp = "{\"problemDetails\": {\t\"type\": \"Bad Request\",\t\"title\": \"N/A\",\t\"status\": 400,\t\"detail\": \"Wrong headers\",\t\"instance\": \"N/A\"}}" self.protocol_version = self.request_version self.send_response(400, 'Bad Request') self.send_header('Host', self.headers.get('Host')) self.send_header('Authorization', self.headers.get('Authorization')) self.send_header('Content-Type', 'application/problem+json') l = len(resp) self.send_header('Content-Length', str(l)) self.end_headers() self.wfile.write(bytes(resp, 'utf-8')) return resp, content_type = self.__process__() print('do_POST: resp= ', resp, ', Content_Type= ', content_type) if (resp == None): self.send_response(404, 'Not Found') self.send_header('Host', self.headers.get('Host')) self.send_header('Authorization', self.headers.get('Authorization')) self.send_header('Content-Type', content_type) self.send_header('Content-Length', '0') self.end_headers() else: if resp.find('Not Found') != -1: self.send_response(404, 'Not Found') elif resp.find('Forbidden') != -1: self.send_response(403, 'Forbidden') elif resp.find('Bad Request') != -1: self.send_response(400, 'Bad Request') elif resp.find('userTrackingSubscription') != -1: self.send_response(201, 'Created') else: self.send_response(200, 'OK') self.send_header('Host', self.headers.get('Host')) self.send_header('Content-Type', content_type) self.send_header('Authorization', self.headers.get('Authorization')) self.send_header('Content-Type', content_type) l = len(resp) self.send_header('Content-Length', str(l)) self.end_headers() self.wfile.write(bytes(resp, 'utf-8')) # End of do_POST # Check HTTP headers def __check_http_headers__(self): print('>>> __check_http_headers__') result = True host = self.headers.get('Host') if host == None: print("__check_http_headers__: Failed to parse Host header") result = False content_type = self.headers.get('Content-Type') if content_type != 'application/json': print("__check_http_headers__: Failed to parse 'Content-Type' header") result = False auth = self.headers.get('Authorization') if auth == None: print("__check_http_headers__: Failed to parse 'Authorization' header") result = False return result # End of __check_http_headers__ # Process the request def __process__(self): s = self.path.split('/') print('>>> __process__: ', s) if s[1] != 'exampleAPI': return None, 'application/json' # TODO Check HTTP headers if s[2] == 'location' and ((s[3] == 'v1') or (s[3] == 'v2')): return self.__process__location__api__(s) elif s[2] == 'ui' and ((s[3] == 'v1') or (s[3] == 'v2')): return self.__process__ue__identity__api__(s) elif s[2] == 'rni' and ((s[3] == 'v1') or (s[3] == 'v2')): return self.__process__rnis__api__(s) elif s[2] == 'bwm' and ((s[3] == 'v1') or (s[3] == 'v2')): return self.__process__bwm__api__(s) else: return None, 'application/json' # End of __process__ # Process the LocationAPI request def __process__location__api__(self, p_split): print (">>> __process__location__api__: ", p_split) resp = None content_type = 'application/json' if p_split[4].startswith('users'): # E.g. users?zoneId=zone01 s = p_split[4].split('?') if s.__len__() == 1: # Users list requested resp = self.__db__.getSubscriberList(self.path) elif s.__len__() == 2: # E.g. zoneId=zone01 s = s[1].split('=') if s[0] == 'zoneId': # Users list into a specific zoneId requested resp = self.__db__.getSubscriberFromZoneId(s[1], self.path) elif s[0] == 'address': resp = self.__db__.getSubscriberFromAddress(s[1]) elif p_split[4] == 'zones': if p_split.__len__() == 5: s = p_split[4].split('?') print("__process__location__api__: ", s) if s.__len__() == 1: # A list requested if s[0] == 'zones': resp = self.__db__.getZoneList(self.path) else: resp = "{\"Unsupported zone request\"}" elif s.__len__() == 2: # E.g. ? resp = "{\"Unsupported zone request\"}" else: resp = "{\"Unsupported " + p_split[4] + " request\"}" elif p_split.__len__() == 7: if p_split[6] == 'accessPoints': resp = self.__db__.getAccessPointList(p_split[5], self.path) else: resp = "{\"Unsupported " + p_split[5] + " request\"}" else: resp = "{\"Unsupported " + p_split[4] + " request\"}" elif p_split[4] == 'subscriptions': if p_split.__len__() == 6: # Extract the body if any body = None content_len = int(self.headers.get('Content-Length')) if (content_len != 0): body = self.rfile.read(content_len) json_msg = self.__decode__json__body__(body) if (json_msg == None): resp = "{\"problemDetails\": {\t\"type\": \"Body processing not supported\",\t\"title\": \"UElocationAPI\",\t\"status\": 400,\t\"detail\": \"Unknown request\",\t\"instance\": \"string\"}}" else: resp = self.__db__.registerSubscription(json_msg, self.path) elif p_split.__len__() == 7: resp = self.__db__.unregisterSubscription(p_split[6]) else: resp = "{\"Unsupported " + p_split[4] + " request\"}" print ("<<< __process__location__api__: ", resp, ", ", content_type) return resp, content_type # End of __process__location__api__ # Process the ue_identityAPI request def __process__ue__identity__api__(self, p_split): print (">>> __process__ue__identity__api__: ", p_split) resp = None content_type = 'application/json' if p_split[4] == 'appInst98': # Used for unknown application, e.g. TC_MEC_PLAT_UETAG_003_BI resp = "{\"problemDetails\": {\t\"type\": \"Not Authorized\",\t\"title\": \"UEidentityAPI\",\t\"status\": 403,\t\"detail\": \"Forbidden\",\t\"instance\": \"AppInst98\"}}" else: if p_split[5].startswith('ue_identity_tag_info'): # E.g. ue_identity_tag_info?ueIdentityTag=UeTagA s = p_split[5].split('?') if s[0] != 'ue_identity_tag_info': resp = "{\"problemDetails\": {\t\"type\": \"Bad Request\",\t\"title\": \"UEidentityAPI\",\t\"status\": 400,\t\"detail\": \"Wrong parameters\",\t\"instance\": \"string\"}}" elif s.__len__() == 1: # ue_identity_tag_info, POST # Register/Unregister operation body = None # Extract the body if any content_len = int(self.headers.get('Content-Length')) if (content_len != 0): body = self.rfile.read(content_len) json_msg = self.__decode__json__body__(body) if (json_msg == None): resp = "{\"problemDetails\": {\t\"type\": \"Body processing not supported\",\t\"title\": \"UEidentityAPI\",\t\"status\": 400,\t\"detail\": \"Unknown request\",\t\"instance\": \"string\"}}" else: resp = self.__db__.registerUEidentity(p_split[4], json_msg) elif s.__len__() == 2: # E.g. ueIdentityTag=UeTagA s = s[1].split('=') if s[0] == 'ueIdentityTag': # Users list into a specific zoneId requested resp = self.__db__.getUEidentityTagInfo(p_split[4], s[1]) if (resp == None): resp = "{\"problemDetails\": {\t\"type\": \"Not Found\",\t\"title\": \"UEidentityAPI\",\t\"status\": 404,\t\"detail\": \"Wrong parameters\",\t\"instance\": \"string\"}}" content_type = 'application/problem+json' if (resp == None): resp = "{\"problemDetails\": {\t\"type\": \"Bad Request\",\t\"title\": \"UEidentityAPI\",\t\"status\": 400,\t\"detail\": \"Wrong parameters\",\t\"instance\": \"string\"}}" content_type = 'application/problem+json' elif resp.find('problemDetails') != -1: content_type = 'application/problem+json' print ("<<< __process__ue__identity__api__: ", resp, ", ", content_type) return resp, content_type # End of __process__ue__identity__api__ # Process the RnisAPI request def __process__rnis__api__(self, p_split): print ('>>> __process__rnis__api__: ', p_split) resp = None content_type = 'application/json' if p_split[4].startswith('subscriptions'): if p_split.__len__() == 6: if p_split[5] == '': resp = self.__db__.getSubscriptionLinkList(self.path) if (resp == None): resp = "{\"problemDetails\": {\t\"type\": \"Not supported\",\t\"title\": \"RnisAPI\",\t\"status\": 400,\t\"detail\": \"Wrong parameters\",\t\"instance\": \"string\"}}" content_type = 'application/problem+json' print ("<<< __process__rnis__api__: ", resp, ", ", content_type) return resp, content_type # End of __process__rnis__api__ # Process the BwManagementAPI request def __process__bwm__api__(self, p_split): print (">>> __process__bwm__api__: ", p_split) resp = None content_type = 'application/json' if p_split[4].startswith('bw_allocations'): # E.g. bw_allocation?app_instance_id=InstApp01 s = p_split[4].split('?') if s.__len__() == 1: # ue_identity_tag_info, POST # Register/Unregister operation body = None # Extract the body if any content_len = int(self.headers.get('Content-Length')) if (content_len != 0): body = self.rfile.read(content_len) json_msg = self.__decode__json__body__(body) if (json_msg == None): resp = "{\"problemDetails\": {\t\"type\": \"Body processing not supported\",\t\"title\": \"BwManagementAPI\",\t\"status\": 400,\t\"detail\": \"Unknown request\",\t\"instance\": \"string\"}}" content_type = 'application/problem+json' else: resp = self.__db__.bwAllocation(json_msg) elif s.__len__() == 2: # E.g. app_instance_id=InstApp01 s = s[1].split('=') if s[0] == 'app_instance_id': resp = self.__db__.getBwAllocationAppInst(p_split[4], s[1]) if (resp == None): resp = "{\"problemDetails\": {\t\"type\": \"Not supported\",\t\"title\": \"BwManagementAPI\",\t\"status\": 400,\t\"detail\": \"Wrong parameters\",\t\"instance\": \"string\"}}" content_type = 'application/problem+json' print ("<<< __process__bwm__api__: ", resp, ", ", content_type) return resp, content_type # End of __process__bwm__api__ # Decode a Json HTTP body message def __decode__json__body__(self, p_body): print(">>> __decode__json__body__: ", p_body) json_msg = json.loads(p_body) print("__decode__json__body__: ", type(json_msg)) print("__decode__json__body__: ", json_msg.keys()) if ("ueIdentityTags" in json_msg.keys()) == True: return json_msg.get("ueIdentityTags")[0] elif ("userTrackingSubscription" in json_msg.keys()) == True: return json_msg.get("userTrackingSubscription") return None # End of __decode__json__body__ # End of class SimpleHTTPRequestHandler