From 0ac0bc155c810cff47b414c8cb8bf443b4e4e246 Mon Sep 17 00:00:00 2001 From: Fawwaz Anugrah Wiradhika Dharmasatya <anugrahdwfawwaz@gmail.com> Date: Mon, 10 Jun 2024 12:01:50 +0700 Subject: [PATCH] refactor: change route terminology to service --- exploration/treesitter.py | 4 +-- src/lib/ACLAnalyzer.py | 26 ++++++++--------- src/lib/ACReader.py | 14 +++++----- src/lib/FileReader.py | 28 +++++++++---------- src/lib/MainMenu.py | 8 +++--- ...yzer.py => ServiceSanitizationAnalyzer.py} | 4 +-- src/lib/VulnerabilitReporter.py | 14 +++++----- tests/tc1/FlaskMainApp.py | 2 +- tests/tc1/auth.py | 8 +++--- tests/tc1/class_views.py | 12 ++++---- tests/tc1/views.py | 12 ++++---- todo.txt | 16 +++++------ 12 files changed, 74 insertions(+), 74 deletions(-) rename src/lib/{RouteSanitizationAnalyzer.py => ServiceSanitizationAnalyzer.py} (99%) diff --git a/exploration/treesitter.py b/exploration/treesitter.py index ba640d9..4441d06 100644 --- a/exploration/treesitter.py +++ b/exploration/treesitter.py @@ -111,7 +111,7 @@ tree = parser.parse( @ACL from flask_login import login_required, current_user -@views.route('/logs', methods=['GET']) +@views.service('/logs', methods=['GET']) @login_required def get_logs(): if (not current_user.admin): @@ -128,7 +128,7 @@ print(tree.root_node.sexp()) STRING = """class ProjectInfo(): def __init__(self) -> None: self.acl_class:Optional[ElementContext] = None - self.route_class:list[ElementContext] = [] + self.service_class:list[ElementContext] = [] @dataclass class ElementContext(): diff --git a/src/lib/ACLAnalyzer.py b/src/lib/ACLAnalyzer.py index c6aa88c..2980119 100644 --- a/src/lib/ACLAnalyzer.py +++ b/src/lib/ACLAnalyzer.py @@ -12,24 +12,24 @@ class ACLAnalyzer(): def analyze(self)->ACLData: real_acl:ACLData = ACLData(self.acl_info.principal_list,{},self.acl_info.acl_context) - for route in self.project_info.route_class: + for route in self.project_info.service_class: format_log(f"Searching for {route.get_base_element_name()} in {route.location}...") # Tentukan tipenya if(route.type == 'function'): - route_acl = self.analyze_function(route) + service_acl = self.analyze_function(route) extra = "" for k in route.context: if k.startswith("Parent::"): extra = f"{k.split('::')[1]}." - real_acl.route_acl[f"{extra}{route.get_base_element_name()}"] = route_acl + real_acl.service_acl[f"{extra}{route.get_base_element_name()}"] = service_acl elif(route.type=='class'): - route_acls = self.analyze_class(route) - for route in route_acls: - real_acl.route_acl[route[0]] = route[1] + service_acls = self.analyze_class(route) + for route in service_acls: + real_acl.service_acl[route[0]] = route[1] elif(route.type=='module'): - route_acls = self.analyze_module(route) - for route in route_acls: - real_acl.route_acl[route[0]] = route[1] + service_acls = self.analyze_module(route) + for route in service_acls: + real_acl.service_acl[route[0]] = route[1] return real_acl @@ -396,20 +396,20 @@ class ACLAnalyzer(): if(met['name']==exempted or (met['type']=='class_method' and met['parent']==exempted)): method_lists = [m for m in method_lists if m['name']!=met['name']] # Cek setiap fungsi - route_acls = [] + service_acls = [] for f in method_lists: if f['type']=='class_method': #TODO class method ctx = ElementContext(f"{f['parent']}.{f['name']}",'function',route.location,[]) ctx.set_cfg(CFG(f['ast'],f['ctx'],f['cursor'],route.cfg.source_code)) result = self.analyze_function(ctx) - route_acls.append([ctx.get_base_element_name(),result]) + service_acls.append([ctx.get_base_element_name(),result]) else: ctx = ElementContext(f['name'],'function',route.location,[]) ctx.set_cfg(CFG(f['ast'],f['ctx'],f['cursor'],route.cfg.source_code)) result = self.analyze_function(ctx) - route_acls.append([ctx.get_base_element_name(),result]) - return route_acls + service_acls.append([ctx.get_base_element_name(),result]) + return service_acls def check_acl_list(self,route:ElementContext,name:str,ctx:list[str]=[])->list[str]|None: #TODO handle variabel dan kelas dan import diff --git a/src/lib/ACReader.py b/src/lib/ACReader.py index fd27005..95e3a37 100644 --- a/src/lib/ACReader.py +++ b/src/lib/ACReader.py @@ -3,13 +3,13 @@ from typing import Optional,Dict,Any,Type import re class ACLData(): - def __init__(self,principal_list:list[str],route_acl:dict[str,list[str]],acl_context:tuple[dict[str,list|dict]]) -> None: + def __init__(self,principal_list:list[str],service_acl:dict[str,list[str]],acl_context:tuple[dict[str,list|dict]]) -> None: self.principal_list:list[str] = principal_list - self.route_acl:dict[str,list[str]] = route_acl + self.service_acl:dict[str,list[str]] = service_acl self.acl_context:tuple[dict[str,list|dict]] = acl_context def __str__(self): - return f"ACLData(principals={self.principal_list},acl={self.route_acl},context={self.acl_context})" + return f"ACLData(principals={self.principal_list},acl={self.service_acl},context={self.acl_context})" def __repr__(self): return self.__str__() @@ -22,7 +22,7 @@ class ACReader(): def read(self) -> ACLData: format_log("Reading ACL File...") principal_list:list[str] = [] - route_acl:dict[str,list[str]] = {} + service_acl:dict[str,list[str]] = {} acl_context:dict[str,list|dict] = {} # Baca File ACL with open(self.acl_path) as acl_file: @@ -57,9 +57,9 @@ class ACReader(): is_all = True break if is_all: - route_acl[route[0]] = [principal for principal in principal_list] + service_acl[route[0]] = [principal for principal in principal_list] else: - route_acl[route[0]] = route[1].split(",") + service_acl[route[0]] = route[1].split(",") elif line_type=='detail': # Parse detail processed_line = line.strip().replace(" ","") @@ -83,5 +83,5 @@ class ACReader(): else: content = detail[1].replace(" ","").split(",") acl_context[detail[0]] = content - acl_data = ACLData(principal_list,route_acl,acl_context) + acl_data = ACLData(principal_list,service_acl,acl_context) return acl_data \ No newline at end of file diff --git a/src/lib/FileReader.py b/src/lib/FileReader.py index 1d3b847..3469117 100644 --- a/src/lib/FileReader.py +++ b/src/lib/FileReader.py @@ -74,7 +74,7 @@ class FileReader: while not is_skipped and re.search(FileReader.ANNOTATION_PATTERN,lines[i]): next_annotations = self.have_annotation(lines[i]) for annon in next_annotations: - if "@Routes" in annon: + if "@Service" in annon: # Skip fungsi ini is_skipped = True break @@ -98,7 +98,7 @@ class FileReader: if parent.type=='block': parent = parent.parent # Kasih keterangan tambahan ke konteks modul/fungsi yang merupakan parent fungsi ini. - for route in project_info.route_class: + for route in project_info.service_class: tmp_parent = parent if tmp_parent.type=='class_definition' and route.type=='module': while tmp_parent.type != 'module': @@ -142,7 +142,7 @@ class FileReader: parent = parent.parent if parent and parent.type=="class_definition": project_info.acl_class[-1].context.append(f"Parent::{parent.children[1].text.decode()}") - elif("@Routes" in annotation): + elif("@Service" in annotation): # Ini kelas routes # Cek jika ada keterangan tambahan match = re.search(r'\((.*?)\)',lines[i]) @@ -164,23 +164,23 @@ class FileReader: continue # Kalau next line nya class, ya class, kalau def yg fungsi, selain itu type nya module if re.search("\S*class +",lines[i]): - project_info.route_class.append(ElementContext(lines[i].strip().split(" ")[1],'class',file,additional_context)) + project_info.service_class.append(ElementContext(lines[i].strip().split(" ")[1],'class',file,additional_context)) elif re.search("\S*def +",lines[i]): - project_info.route_class.append(ElementContext(lines[i].strip().split(" ")[1],'function',file,additional_context)) + project_info.service_class.append(ElementContext(lines[i].strip().split(" ")[1],'function',file,additional_context)) else: # Dapatkan konteks berupa cfg setiap kelas - project_info.route_class.append(ElementContext(".".join(file.split(PATH_SEPARATOR)[-1].split(".")[:-1]),'module',file,additional_context)) + project_info.service_class.append(ElementContext(".".join(file.split(PATH_SEPARATOR)[-1].split(".")[:-1]),'module',file,additional_context)) # Generate CFG jika ini rute - cfg,deps = self.cfg_generator.generate(project_info.route_class[-1]) - project_info.route_class[-1].set_cfg(cfg) + cfg,deps = self.cfg_generator.generate(project_info.service_class[-1]) + project_info.service_class[-1].set_cfg(cfg) project_info.dependency_manager.add(deps) # Cek jika ini fungsi dalam kelas - if(project_info.route_class[-1].type=='function'): - parent = project_info.route_class[-1].cfg.ast.parent.parent - if project_info.route_class[-1].cfg.ast.parent.type=='decorated_definition': + if(project_info.service_class[-1].type=='function'): + parent = project_info.service_class[-1].cfg.ast.parent.parent + if project_info.service_class[-1].cfg.ast.parent.type=='decorated_definition': parent = parent.parent if parent and parent.type=="class_definition": - project_info.route_class[-1].context.append(f"Parent::{parent.children[1].text.decode()}") + project_info.service_class[-1].context.append(f"Parent::{parent.children[1].text.decode()}") else: # Ini kelas biasa pass @@ -216,12 +216,12 @@ class FileReader: class ProjectInfo(): def __init__(self) -> None: self.acl_class:list[ElementContext] = [] - self.route_class:list[ElementContext] = [] + self.service_class:list[ElementContext] = [] self.dependency_manager:DependencyManager= DependencyManager() def __repr__(self) -> str: return self.__str__() def __str__(self) -> str: - return f"ProjectInfo(acl_class={self.acl_class},route_class={self.route_class},dependency_manager={self.dependency_manager})" + return f"ProjectInfo(acl_class={self.acl_class},service_class={self.service_class},dependency_manager={self.dependency_manager})" \ No newline at end of file diff --git a/src/lib/MainMenu.py b/src/lib/MainMenu.py index 7692a18..c4d191e 100644 --- a/src/lib/MainMenu.py +++ b/src/lib/MainMenu.py @@ -1,7 +1,7 @@ from lib.utils import format_log from lib.ACReader import ACReader,ACLData from lib.FileReader import FileReader -from lib.RouteSanitizationAnalyzer import RouteSanitizationAnalyzer +from lib.ServiceSanitizationAnalyzer import ServiceSanitizationAnalyzer from lib.ACLAnalyzer import ACLAnalyzer from lib.VulnerabilitReporter import VulnerabilityReporter import time @@ -18,7 +18,7 @@ class MainMenu(): def __init__(self): self.project_path:str = "" self.acl_path:str = "" - # @Routes + # @Service def start(self): self.print_header() # Dapatkan Path Menuju File dan Konfigurasi ACL @@ -46,7 +46,7 @@ class MainMenu(): reporter = VulnerabilityReporter(self.acl_data) # Analisis sanitasi router format_log("Starting route sanitization analysis...") - reporter.add_unsanitized_routes(RouteSanitizationAnalyzer().set_project(self.project_ctx).analyze()) + reporter.add_unsanitized_routes(ServiceSanitizationAnalyzer().set_project(self.project_ctx).analyze()) # Analisis kontrol akses format_log("Starting access control analysis...") reporter.add_acl_data(ACLAnalyzer(self.project_ctx,self.acl_data).analyze()) @@ -56,7 +56,7 @@ class MainMenu(): reporter.get_acl_table() finally: format_log(f"Analysis finished in {time.time()-start_time} seconds.",end="\n\n") - # @Routes + # @Service @explore_decorator_no_args # @explore_decorator def print_header(self): diff --git a/src/lib/RouteSanitizationAnalyzer.py b/src/lib/ServiceSanitizationAnalyzer.py similarity index 99% rename from src/lib/RouteSanitizationAnalyzer.py rename to src/lib/ServiceSanitizationAnalyzer.py index 325e1e3..358608f 100644 --- a/src/lib/RouteSanitizationAnalyzer.py +++ b/src/lib/ServiceSanitizationAnalyzer.py @@ -4,13 +4,13 @@ from lib.utils import format_log,flatten_node from .CFGGenerator import CFG from tree_sitter import Node from .Const import PATH_SEPARATOR -class RouteSanitizationAnalyzer(): +class ServiceSanitizationAnalyzer(): def set_project(self,project_info:ProjectInfo): self.project_info = project_info return self def analyze(self)->list[str]: vuln_routes = [] - for route in self.project_info.route_class: + for route in self.project_info.service_class: format_log(f"Analyzing {route.get_base_element_name()} in {route.location}...") # Tentukan tipenya if(route.type == 'function'): diff --git a/src/lib/VulnerabilitReporter.py b/src/lib/VulnerabilitReporter.py index 8d9d685..1fcd822 100644 --- a/src/lib/VulnerabilitReporter.py +++ b/src/lib/VulnerabilitReporter.py @@ -17,7 +17,7 @@ class VulnerabilityReporter: extra = "" try: key = self.element_not_contacting_acl[i].split("(")[0] - if self.expected_acl_data.route_acl[key]==self.real_acl_data.principal_list: + if self.expected_acl_data.service_acl[key]==self.real_acl_data.principal_list: extra = " | Possibily doesn't need control access?" except KeyError: pass @@ -26,22 +26,22 @@ class VulnerabilityReporter: def get_acl_table(self)->None: print("> Real ACL Table:") - if not self.real_acl_data.route_acl: + if not self.real_acl_data.service_acl: print("-") else: - for key,value in self.real_acl_data.route_acl.items(): + for key,value in self.real_acl_data.service_acl.items(): if value =='*': value = ",".join(self.real_acl_data.principal_list) print(f"{key}: {value}") print() print("> Potential Broken ACL:") i = 0 - for key,value in self.real_acl_data.route_acl.items(): + for key,value in self.real_acl_data.service_acl.items(): try: - if set(self.expected_acl_data.route_acl[key])!=set(self.real_acl_data.route_acl[key]): + if set(self.expected_acl_data.service_acl[key])!=set(self.real_acl_data.service_acl[key]): # Discrepancy - set_expected = set(self.expected_acl_data.route_acl[key]) - set_real = set(self.real_acl_data.route_acl[key]) + set_expected = set(self.expected_acl_data.service_acl[key]) + set_real = set(self.real_acl_data.service_acl[key]) discrepancy = list((set_expected - set_real).union(set_real - set_expected)) print(f"{i+1}.{key} | Principal list difference: [{','.join(discrepancy)}]") i+=1 diff --git a/tests/tc1/FlaskMainApp.py b/tests/tc1/FlaskMainApp.py index a05ddde..37ca0e9 100644 --- a/tests/tc1/FlaskMainApp.py +++ b/tests/tc1/FlaskMainApp.py @@ -15,7 +15,7 @@ def create_app(): app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{DB_NAME}' db.init_app(app) - # register routes + # register services register_controllers(app) #set config diff --git a/tests/tc1/auth.py b/tests/tc1/auth.py index eb316f9..2ac01a1 100644 --- a/tests/tc1/auth.py +++ b/tests/tc1/auth.py @@ -8,7 +8,7 @@ from flask_login import login_user, login_required, logout_user, current_user auth = Blueprint('auth', __name__) -@auth.route('/login', methods=['GET', 'POST']) +@auth.service('/login', methods=['GET', 'POST']) def login(): if request.method == 'POST': email = request.form.get('email') @@ -27,15 +27,15 @@ def login(): return render_template("login.html") -#@Routes -@auth.route('/logout') +#@Service +@auth.service('/logout') @login_required def logout(): logout_user() return redirect(url_for('auth.login')) -@auth.route('/sign-up', methods=['GET', 'POST']) +@auth.service('/sign-up', methods=['GET', 'POST']) def sign_up(): if request.method == 'POST': email = request.form.get('email') diff --git a/tests/tc1/class_views.py b/tests/tc1/class_views.py index 348327a..f41dccb 100644 --- a/tests/tc1/class_views.py +++ b/tests/tc1/class_views.py @@ -5,12 +5,12 @@ from .models import Note,Log from .db import db import json from lib.RoleCheck import RoleCheck -# @Routes +# @Service views = Blueprint('views', __name__) class Views(): def __init__(self,a) -> None: self.a = a - @views.route('/add', methods=['POST']) + @views.service('/add', methods=['POST']) @login_required def class_add_note(self): if request.method == 'POST': @@ -26,7 +26,7 @@ class Views(): return render_template("home.html") - @views.route('/update', methods=['POST']) + @views.service('/update', methods=['POST']) @login_required def class_update_note(self): if request.method == 'POST': @@ -44,7 +44,7 @@ class Views(): return render_template("home.html") - @views.route('/delete', methods=['POST']) + @views.service('/delete', methods=['POST']) @login_required def class_delete_note(self): note = json.loads(request.data) @@ -56,12 +56,12 @@ class Views(): db.session.commit() return jsonify({}) - @views.route('/note', methods=['GET']) + @views.service('/note', methods=['GET']) @login_required def class_get_note(self): notes = Note.query.filter(Note.user_id==current_user.id) return jsonify(notes) - @views.route('/logs', methods=['GET']) + @views.service('/logs', methods=['GET']) # @login_required def class_get_logs(self): if (RoleCheck().is_admin(current_user)): diff --git a/tests/tc1/views.py b/tests/tc1/views.py index 86effa9..7e761a9 100644 --- a/tests/tc1/views.py +++ b/tests/tc1/views.py @@ -1,4 +1,4 @@ -# @Routes +# @Service from flask import Blueprint, render_template, request, flash, jsonify,abort # @A\CL(login_required) from flask_login import login_required, current_user @@ -10,7 +10,7 @@ from lib.RoleCheck import RoleCheck views = Blueprint('views', __name__) -@views.route('/add', methods=['POST']) +@views.service('/add', methods=['POST']) @login_required def add_note(): if request.method == 'POST': @@ -26,7 +26,7 @@ def add_note(): return render_template("home.html") -@views.route('/update', methods=['POST']) +@views.service('/update', methods=['POST']) @login_required def update_note(): if request.method == 'POST': @@ -46,7 +46,7 @@ def update_note(): return render_template("home.html") -@views.route('/delete', methods=['POST']) +@views.service('/delete', methods=['POST']) @login_required def delete_note(): note = json.loads(request.data) @@ -59,13 +59,13 @@ def delete_note(): return jsonify({}) -@views.route('/note', methods=['GET']) +@views.service('/note', methods=['GET']) @login_required def get_note(): notes = Note.query.filter(Note.user_id==current_user.id) return jsonify(notes) -@views.route('/logs', methods=['GET']) +@views.service('/logs', methods=['GET']) # @login_required def get_logs(): rolecheck = RoleCheck() diff --git a/todo.txt b/todo.txt index c046334..9ff6afd 100644 --- a/todo.txt +++ b/todo.txt @@ -6,8 +6,8 @@ module: - function (V) - class (V) - class > function (V) -- routes kena nocheck? -routes -> nocheck +- services kena nocheck? +services -> nocheck function (V) class: (V) - function (V) @@ -15,7 +15,7 @@ module: (V) - function (V) - class (V) - class > function (V) -nocheck -> routes +nocheck -> services function (V) class:(V) - function (V) @@ -27,15 +27,15 @@ module: (V) TODO - check kalau kelasnya disimpan di variabel (V) -> cuma kalau langsung manggil kelas ( 1 kali assignment, belum transitif) -- acl sekaligus routes? -routes -> acl -acl -> routes +- acl sekaligus services? +services -> acl +acl -> services nocheck -> acl acl -> nocheck -routes -> routes +services -> services nocheck -> nocheck acl -> acl -- routes dalam routes (module -> function) +- services dalam services (module -> function) - handle remaining duplicate annotation TODO: handle path berspasi -- GitLab