!104 Add an acl check script

Merge pull request !104 from 吴留栓/wls
This commit is contained in:
openharmony_ci
2023-06-20 01:45:22 +00:00
committed by Gitee
6 changed files with 422 additions and 0 deletions
@@ -0,0 +1,75 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2023 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import sys
sys.path.append(os.path.dirname(os.path.realpath(__file__)) + os.sep)
from resolve_token_info import *
from read_acl_whitelist import *
log_tag = 'acl_check'
def whitelist_check(whitelist, acls):
try:
set_log_content(LogLevel(2).name, log_tag + '->whitelist_check',
'-------------------------- Trustlist Verification begin --------------------------')
check_pass = True
for k, v in acls.items():
if k in whitelist.keys():
temp = whitelist[k]
for acl in v:
if acl not in temp:
check_pass = False
set_log_content(LogLevel(2).name, log_tag, log_tag + '->whitelist_check',
'precessName = {} the acl = {} trustlist is not configured.'.format(k, acl))
else:
check_pass = False
set_log_content(LogLevel(2).name, log_tag + '->whitelist_check', 'precessName = {} the acls = {} trustlist is not configured.'.format(k, v))
if check_pass == False:
raise AclCheckException(
'-------------------------- Trustlist Verification failed --------------------------')
else:
set_log_content(LogLevel(2).name, log_tag + '->whitelist_check',
'-------------------------- Trustlist Verification successful --------------------------')
except Exception as e:
set_log_content(LogLevel(1).name, log_tag + '->whitelist_check', e.msg)
raise
def main(sn):
set_log_content(LogLevel(2).name, log_tag,
'-------------------------- ACL check begin --------------------------')
try:
hdc_command(GENERATING_TOKEN_INFO_COMMAND.format(sn, TOKEN_INFO_URL))
hdc_command(DOWNLOAD_TOKEN_INFO_COMMAND.format(sn, TOKEN_INFO_URL, DOWNLOAD_TOKEN_INFO_URL))
hdc_command(CLEAR_TOKEN_INFO_FILE.format(sn, TOKEN_INFO_URL))
file = read_txt(DOWNLOAD_TOKEN_INFO_URL)
clear_token_info_txt(DOWNLOAD_TOKEN_INFO_URL)
acls_dict = check_and_get(file)
acl_whitelist = read_json(PATH + 'acl_whitelist.json')
whitelist = get_acl_dict(acl_whitelist)
whitelist_check(whitelist, acls_dict)
except Exception as e:
set_log_content(LogLevel(1).name, log_tag, e.msg)
set_log_content(LogLevel(1).name, log_tag,
'-------------------------- ACL check failed --------------------------')
finally:
set_log_content(LogLevel(2).name, log_tag,
'-------------------------- ACL check end --------------------------')
if __name__ == '__main__':
sn = sys.argv[1]
main(sn)
@@ -0,0 +1,118 @@
[
{
"processName": "hiview",
"acls": [
"ohos.permission.DUMP"
]
},
{
"processName": "privacy_service",
"acls": [
"ohos.permission.MANAGE_DISPOSED_APP_STATUS"
]
},
{
"processName": "inputmethod_service",
"acls": [
"ohos.permission.INPUT_MONITORING"
]
},
{
"processName": "memmgrservice",
"acls": [
"ohos.permission.INTERACT_ACROSS_LOCAL_ACCOUNTS_EXTENSION"
]
},
{
"processName": "locationhub",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "useriam",
"acls": [
"ohos.permission.ACCESS_AUTH_RESPOOL",
"ohos.permission.INTERACT_ACROSS_LOCAL_ACCOUNTS_EXTENSION"
]
},
{
"processName": "pinauth",
"acls": [
"ohos.permission.ACCESS_AUTH_RESPOOL"
]
},
{
"processName": "foundation",
"acls": [
"ohos.permission.PUBLISH_SYSTEM_COMMON_EVENT",
"ohos.permission.PERMISSION_START_ABILITIES_FROM_BACKGROUND",
"ohos.permission.GRANT_SENSITIVE_PERMISSIONS",
"ohos.permission.REVOKE_SENSITIVE_PERMISSIONS",
"ohos.permission.INPUT_MONITORING"
]
},
{
"processName": "dscreen",
"acls": [
"ohos.permission.CAPTURE_SCREEN"
]
},
{
"processName": "sensors",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "camera_service",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "audio_policy",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
},
{
"processName": "msdp_sa",
"acls": [
"ohos.permission.INPUT_MONITORING"
]
},
{
"processName": "dslm_service",
"acls": [
"ohos.permission.ACCESS_IDS"
]
},
{
"processName": "accountmgr",
"acls": [
"ohos.permission.ENFORCE_USER_IDM"
]
},
{
"processName": "hdcd",
"acls": [
"ohos.permission.GET_BUNDLE_INFO_PRIVILEGED",
"ohos.permission.INSTALL_BUNDLE",
"ohos.permission.LISTEN_BUNDLE_CHANGE",
"ohos.permission.CHANGE_ABILITY_ENABLED_STATE",
"ohos.permission.REMOVE_CACHE_FILES",
"ohos.permission.START_ABILITIES_FROM_BACKGROUND",
"ohos.permission.PERMISSION_USED_STATS",
"ohos.permission.DUMP",
"ohos.permission.NOTIFICATION_CONTROLLER",
"ohos.permission.PUBLISH_SYSTEM_COMMON_EVENT"
]
},
{
"processName": "softbus_server",
"acls": [
"ohos.permission.GET_SENSITIVE_PERMISSIONS"
]
}
]
@@ -0,0 +1,37 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2023 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import platform
import time
# 系统分隔符
SYS_SEQ = os.sep
# 系统平台
SYS_PLATFORM = platform.system()
PATH = os.path.dirname(os.path.realpath(__file__)) + SYS_SEQ
# 脚本的执行日志
LOG_FILE = PATH + SYS_SEQ + "native_sa.log"
# 设备上生成的token info 文件名
TOKEN_INFO_NAME = 'token_info_{}.txt'.format(time.time_ns())
# 设备上生成文件存放位置
TOKEN_INFO_URL = '/data/{}'.format(TOKEN_INFO_NAME)
# 设备上文件生成命令
GENERATING_TOKEN_INFO_COMMAND = 'hdc -t {} shell atm dump -t > {}'
# 下载token info 文件存放路径
DOWNLOAD_TOKEN_INFO_URL = PATH + TOKEN_INFO_NAME
# 文件下载命令
DOWNLOAD_TOKEN_INFO_COMMAND = 'hdc -t {} file recv {} {}'
# 删除设备上的文件命令
CLEAR_TOKEN_INFO_FILE = 'hdc -t {} shell rm -rf {}'
@@ -0,0 +1,49 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2023 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import sys
sys.path.append(os.path.dirname(os.path.realpath(__file__)) + os.sep)
import json
from utils import *
log_tag = 'read_acl_whitelist'
def read_json(path):
set_log_content(LogLevel(2).name, log_tag, 'read {}'.format(path))
if not os.path.exists(path):
set_log_content(LogLevel(2).name, log_tag, '{} file not exits'.format(path))
raise AclCheckException('{} file not exits'.format(path))
try:
with open(path, 'r') as f:
file = f.read()
return file
except Exception as e:
set_log_content(LogLevel(1).name, log_tag, e.msg)
raise AclCheckException('{} failed to read the file.'.format(path))
def get_acl_dict(file):
try:
acls_dict = {}
f = json.loads(file)
for it in f:
key = it.get('processName')
values = it.get('acls')
acls_dict[key] = values
return acls_dict
except Exception as e:
set_log_content(LogLevel(1).name, log_tag, '{}'.format(e.msg))
raise
@@ -0,0 +1,72 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2023 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import sys
sys.path.append(os.path.dirname(os.path.realpath(__file__)) + os.sep)
from utils import *
log_tag = 'resolve_token_info'
def check_and_get(file: list):
nativeAcls = {}
try:
set_log_content(LogLevel(2).name, log_tag,
'-------------------------- invalidPermList check begin --------------------------')
check_pass = True
processName = 'xxxxxxxx'
for it in file:
if it.find('processName') != -1:
processName = it.replace(',', '').split(':')[1].split('"')[1]
elif it.find('invalidPermList') != -1:
check_pass = False
msg = 'invalidPermList information is detected in processName = {}'.format(processName)
set_log_content(LogLevel(2).name, log_tag, msg)
elif check_pass and it.find('nativeAcls') != -1:
bb = it.split(':')
if bb[1].split('"')[1].__len__() == 0:
continue
permissionNameList = bb[1].split('"')[1].split(',')
nativeAcls[processName] = permissionNameList
if check_pass == False:
raise AclCheckException('-------------------------- The invalidPermList check failed --------------------------')
else:
set_log_content(LogLevel(2).name, log_tag,
'-------------------------- The invalidPermList check successful --------------------------')
except Exception as e:
set_log_content(LogLevel(1).name, log_tag, e.msg)
raise
return nativeAcls
def clear_token_info_txt(path):
try:
os.remove(path)
except Exception as e:
set_log_content(LogLevel(1).name, log_tag, e.msg)
def read_txt(path):
set_log_content(LogLevel(2).name, log_tag, 'read {}'.format(path))
if not os.path.exists(path):
set_log_content(LogLevel(2).name, log_tag, '{} file not exits'.format(path))
raise AclCheckException('{} file not exits!'.format(path))
try:
with open(path, 'r') as f:
file = f.readlines()
return file
except Exception as e:
set_log_content(LogLevel(1).name, log_tag, e.msg)
raise AclCheckException('{} failed to read the file.'.format(path))
@@ -0,0 +1,71 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2023 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import enum
import logging
import os
import sys
from subprocess import Popen, PIPE, STDOUT
sys.path.append(os.path.dirname(os.path.realpath(__file__)) + os.sep)
from config import *
log_tag = 'utils'
class AclCheckException(Exception):
def __init__(self, msg):
self.msg = msg
def timestamp():
return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())
class LogLevel(enum.Enum):
Error = 1
Info = 2
logging.basicConfig(filename=LOG_FILE, level=logging.INFO, format='%(asctime)s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S %a')
def log(msg):
logging.error(msg)
def set_log_content(level, tag, msg):
log_content = timestamp() + ' {}'.format(level) + ' [{}]'.format(tag) + ' {}'.format(msg)
print(log_content)
log(log_content)
return (log_content)
def shell_command(command_list: list):
try:
print(command_list)
process = Popen(command_list, stdout=PIPE, stderr=STDOUT)
exitcode = process.wait()
set_log_content(LogLevel(2).name, log_tag, '{} operation fuccessful!'.format(command_list))
return process, exitcode
except Exception as e:
set_log_content(LogLevel(1).name, log_tag, e.msg)
raise AclCheckException(e.msg)
def hdc_command(command):
print(command)
command_list = command.split(' ')
_, exitcode = shell_command(command_list)
return exitcode