Coverage for kea/app_hm.py: 76%

1import logging 

2import os 

3import hashlib 

4import subprocess, shlex 

5import zipfile 

6import json 

7import shutil 

8 

9from typing import TYPE_CHECKING 

10if TYPE_CHECKING: 

11 from .start import Setting 

12 

13if __name__ != "__main__": 

14 from .intent import Intent 

15 

16def run_cmd(cmd): 

17 return subprocess.run(shlex.split(cmd), shell=True, check=True).stdout 

18 

19class AppHM(object): 

20 """ 

21 this class describes an app 

22 """ 

23 

24 def __init__(self, app_path, output_dir=None, settings:"Setting"=None): 

25 """ 

26 create an App instance 

27 :param app_path: local file path of app 

28 :return: 

29 """ 

30 # assert app_path is not None 

31 self.logger = logging.getLogger(self.__class__.__name__) 

32 

33 self.app_path = app_path 

34 

35 self.output_dir = output_dir 

36 if output_dir is not None: 

37 if not os.path.isdir(output_dir): 

38 os.makedirs(output_dir) 

39 

40 self.settings = settings 

41 

42 if not self.settings.is_package: 

43 self._hap_init() 

44 else: 

45 self._package_init(package_name=app_path) 

46 

47 def _package_init(self, package_name): 

48 self.package_name = package_name 

49 self.main_activity = self._dumpsys_package_info 

50 

51 @property 

52 def _dumpsys_package_info(self): 

53 from .adapter.hdc import HDC_EXEC 

54 cmd = [HDC_EXEC, "-t", self.settings.device_serial, "shell", "bm", "dump", "-n", self.package_name] 

55 r = subprocess.check_output(cmd, text=True) 

56 package_info_str = r.split("\n", maxsplit=1)[-1] 

57 import json 

58 package_info = json.loads(package_info_str) 

59 return package_info["hapModuleInfos"][0]["mainAbility"] 

60 

61 def __str__(self) -> str: 

62 app_info = ", ".join([ 

63 f"bundleName:{self.package_name}", 

64 f"total_abilities:{self.total_activities}", 

65 f"main_ability:{self.main_activity}", 

66 ]) 

67 return f"App({app_info})" 

68 

69 def _hap_init(self): 

70 self.logger.info(f"Extracting info from {self.app_path}") 

71 self.logger.info(f"Hapfile is {os.path.basename(self.app_path)}") 

72 # make temp dir 

73 temp_dir = os.path.join(os.path.dirname(self.app_path), "temp_hap") 

74 # temp_dir = "/".join(self.app_path.split("/")[:-1]) + "/temp_hap" 

75 if os.path.exists(temp_dir): 

76 shutil.rmtree(temp_dir) 

77 os.mkdir(temp_dir) 

78 

79 with zipfile.ZipFile(self.app_path, 'r') as zip_ref: 

80 zip_ref.extractall(temp_dir) 

81 

82 with open(temp_dir + "/module.json") as f: 

83 moudle_json = json.load(f) 

84 

85 with open(temp_dir + "/pack.info") as f: 

86 pack_info = json.load(f) 

87 

88 self.load_hap_info(moudle_json, pack_info) 

89 

90 shutil.rmtree(temp_dir) 

91 

92 def load_hap_info(self, module_json, pack_info): 

93 self.package_name = pack_info["summary"]["app"]["bundleName"] 

94 self.api_version = pack_info["summary"]["modules"][0]["apiVersion"]["target"] 

95 self.main_activity = pack_info["summary"]["modules"][0]["mainAbility"] 

96 self.activities = pack_info["summary"]["modules"][0]["abilities"] 

97 self.total_activities = len(self.activities) 

98 # self.possible_broadcasts = self.get_possible_broadcasts() 

99 self.hashes = self.get_hashes() 

100 

101 def get_package_name(self): 

102 """ 

103 get package name of current app 

104 :return: 

105 """ 

106 return self.package_name 

107 

108 def get_main_activity(self): 

109 """ 

110 get package name of current app 

111 :return: 

112 """ 

113 if self.main_activity is not None: 

114 return self.main_activity 

115 else: 

116 self.logger.warning("Cannot get main activity from manifest.") 

117 # return self.dumpsys_main_activity 

118 

119 def get_start_intent(self): 

120 """ 

121 get an intent to start the app 

122 :return: Intent 

123 """ 

124 bundle_name = self.get_package_name() 

125 main_ability = self.get_main_activity() 

126 # hdc shell aa -b [bundleName] -a [Main ability] 

127 return Intent(suffix="-b {} -a {}".format(bundle_name, main_ability), is_harmonyos=True) 

128 

129 # def get_start_with_profiling_intent(self, trace_file, sampling=None): 

130 # """ 

131 # get an intent to start the app with profiling 

132 # :return: Intent 

133 # """ 

134 # package_name = self.get_package_name() 

135 # if self.get_main_activity(): 

136 # package_name += "/%s" % self.get_main_activity() 

137 # if sampling is not None: 

138 # return Intent(prefix="start --start-profiler %s --sampling %d" % (trace_file, sampling), suffix=package_name) 

139 # else: 

140 # return Intent(prefix="start --start-profiler %s" % trace_file, suffix=package_name) 

141 

142 def get_stop_intent(self): 

143 """ 

144 get an intent to stop the app 

145 :return: Intent 

146 """ 

147 bundle_name = self.get_package_name() 

148 return Intent(prefix="force-stop", suffix=bundle_name, is_harmonyos=True) 

149 

150 # def get_possible_broadcasts(self): 

151 # possible_broadcasts = set() 

152 # for receiver in self.apk.get_receivers(): 

153 # intent_filters = self.apk.get_intent_filters('receiver', receiver) 

154 # actions = intent_filters['action'] if 'action' in intent_filters else [] 

155 # categories = intent_filters['category'] if 'category' in intent_filters else [] 

156 # categories.append(None) 

157 # for action in actions: 

158 # for category in categories: 

159 # intent = Intent(prefix='broadcast', action=action, category=category) 

160 # possible_broadcasts.add(intent) 

161 # return possible_broadcasts 

162 

163 def get_hashes(self, block_size=2 ** 8): 

164 """ 

165 Calculate MD5,SHA-1, SHA-256 

166 hashes of APK input file 

167 @param block_size: 

168 """ 

169 md5 = hashlib.md5() 

170 sha1 = hashlib.sha1() 

171 sha256 = hashlib.sha256() 

172 f = open(self.app_path, 'rb') 

173 while True: 

174 data = f.read(block_size) 

175 if not data: 

176 break 

177 md5.update(data) 

178 sha1.update(data) 

179 sha256.update(data) 

180 return [md5.hexdigest(), sha1.hexdigest(), sha256.hexdigest()] 

181 

182if __name__ == "__main__": 

183 import argparse 

184 parser = argparse.ArgumentParser(description="agr parser for Extrating info from hap") 

185 parser.add_argument("-a", "--app", type=str, help="the address to the .hap file", required=True) 

186 args = parser.parse_args() 

187 app = AppHM(args.app) 

188 # * uncomment the below block to save the ablity info to file 

189 # with open(app.package_name+".info.json", "w") as fp: 

190 # import json 

191 # json.dump(app.activities, fp, indent=4, sort_keys=True) 

192 print(app)