211 lines
7.8 KiB
Python
211 lines
7.8 KiB
Python
# osmo_gsm_tester: specifics for setting an APN on an AndroidUE modem
|
|
#
|
|
# Copyright (C) 2020 by Software Radio Systems Limited
|
|
#
|
|
# Author: Nils Fürste <nils.fuerste@softwareradiosystems.com>
|
|
# Author: Bedran Karakoc <bedran.karakoc@softwareradiosystems.com>
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import re
|
|
|
|
from ..core import log
|
|
from ..core import schema
|
|
from .android_host import AndroidHost
|
|
|
|
|
|
class AndroidApn(AndroidHost):
|
|
##############
|
|
# PROTECTED
|
|
##############
|
|
def __init__(self, apn, mcc, mnc, select=None):
|
|
self.logger_name = 'apn_worker_'
|
|
super().__init__(self.logger_name)
|
|
self._apn_name = apn
|
|
self._apn = apn
|
|
self._mcc = mcc
|
|
self._mnc = mnc
|
|
self._select = select
|
|
if not self._apn:
|
|
raise log.Error('APN name not set')
|
|
if not self._mcc:
|
|
raise log.Error('MCC not set')
|
|
if not self._mnc:
|
|
raise log.Error('MNC not set')
|
|
|
|
# optional parameters, set with set_additional_params()
|
|
self.proxy = None
|
|
self.port = None
|
|
self.user = None
|
|
self.password = None
|
|
self.server = None
|
|
self.mmsc = None
|
|
self.mmsport = None
|
|
self.mmsproxy = None
|
|
self.auth = None
|
|
self.type = None
|
|
self.protocol = None
|
|
self.mvnoval = None
|
|
self.mvnotype = None
|
|
|
|
def get_carrier_id(self, carrier_name):
|
|
qry_carrier_cmd = "content query --uri \"content://telephony/carriers\""
|
|
proc = self.run_androidue_cmd('get-carrier-id', [qry_carrier_cmd])
|
|
proc.launch_sync()
|
|
available_carriers = proc.get_stdout().split('\n')
|
|
carr_id = -1
|
|
for carr in available_carriers:
|
|
if 'name=' + carrier_name in carr: # found carrier
|
|
carr_id = re.findall(r'_id=(\S+),', carr)[0]
|
|
break
|
|
return carr_id
|
|
|
|
def set_new_carrier(self, apn_parameter, carr_id):
|
|
# check if carrier was found, delete it if exists
|
|
if carr_id != -1:
|
|
self.delete_apn(apn_parameter['carrier'])
|
|
|
|
set_carrier_cmd = "content insert --uri content://telephony/carriers" \
|
|
+ " --bind name:s:\"" + apn_parameter["carrier"] + "\"" \
|
|
+ " --bind numeric:s:\"" + apn_parameter["mcc"] + apn_parameter["mnc"] + "\"" \
|
|
+ " --bind mcc:s:\"" + apn_parameter["mcc"] + "\"" \
|
|
+ " --bind mnc:s:\"" + apn_parameter["mnc"] + "\""\
|
|
+ " --bind apn:s:\"" + apn_parameter["apn"] + "\"" \
|
|
+ " --bind user:s:\"" + apn_parameter["user"] + "\"" \
|
|
+ " --bind password:s:\"" + apn_parameter["password"] + "\"" \
|
|
+ " --bind mmsc:s:\"" + apn_parameter["mmsc"] + "\"" \
|
|
+ " --bind mmsport:s:\"" + apn_parameter["mmsport"] + "\"" \
|
|
+ " --bind mmsproxy:s:\"" + apn_parameter["mmsproxy"] + "\"" \
|
|
+ " --bind authtype:s:\"" + apn_parameter["auth"] + "\"" \
|
|
+ " --bind type:s:\"" + apn_parameter["type"] + "\"" \
|
|
+ " --bind protocol:s:\"" + apn_parameter["protocol"] + "\"" \
|
|
+ " --bind mvno_type:s:\"" + apn_parameter["mvnotype"] + "\"" \
|
|
+ " --bind mvno_match_data:s:\"" + apn_parameter["mvnoval"] + "\""
|
|
proc = self.run_androidue_cmd("set-new-carrier", [set_carrier_cmd])
|
|
proc.launch_sync()
|
|
return self.get_carrier_id(apn_parameter['carrier'])
|
|
|
|
def set_preferred_apn(self, carr_id):
|
|
if carr_id != -1:
|
|
set_apn_cmd = "content insert --uri content://telephony/carriers/preferapn --bind apn_id:s:\"" + str(carr_id) + "\""
|
|
proc = self.run_androidue_cmd('set-preferred-apn', [set_apn_cmd])
|
|
proc.launch_sync()
|
|
|
|
def select_apn(self, carr_name):
|
|
carr_id = self.get_carrier_id(carr_name)
|
|
if carr_id == 0:
|
|
return False
|
|
|
|
# select carrier by ID
|
|
sel_apn_cmd = "content update --uri content://telephony/carriers/preferapn --bind apn_id:s:\"" + str(carr_id) + "\""
|
|
proc = self.run_androidue_cmd('select-apn', [sel_apn_cmd])
|
|
proc.launch_sync()
|
|
return True
|
|
|
|
def delete_apn(self, carr_name):
|
|
set_apn_cmd = "content delete --uri content://telephony/carriers --where \'name=\"" + str(carr_name) + "\" \'"
|
|
proc = self.run_androidue_cmd('delete-apn', [set_apn_cmd])
|
|
proc.launch_sync()
|
|
|
|
########################
|
|
# PUBLIC - INTERNAL API
|
|
########################
|
|
@classmethod
|
|
def from_conf(cls, conf):
|
|
return cls(conf.get('apn', None), conf.get('mcc', None),
|
|
conf.get('mnc', None), conf.get('select', None))
|
|
|
|
@classmethod
|
|
def schema(cls):
|
|
resource_schema = {
|
|
'apn': schema.STR,
|
|
'mcc': schema.STR,
|
|
'mnc': schema.STR,
|
|
'select': schema.BOOL_STR,
|
|
}
|
|
return resource_schema
|
|
|
|
def configure(self, testenv, run_dir, run_node, rem_host):
|
|
self.testenv = testenv
|
|
self.rem_host = rem_host
|
|
self._run_node = run_node
|
|
self.run_dir = run_dir
|
|
self.logger_name += self._run_node.run_addr()
|
|
self.set_name(self.logger_name)
|
|
|
|
def set_additional_params(self, proxy=None, port=None, user=None, password=None, server=None, auth=None, apn_type=None,
|
|
mmsc=None, mmsport=None, mmsproxy=None, protocol=None, mvnoval=None, mvnotype=None):
|
|
self.proxy = proxy
|
|
self.port = port
|
|
self.user = user
|
|
self.password = password
|
|
self.server = server
|
|
self.auth = auth
|
|
self.type = apn_type
|
|
self.mmsc = mmsc
|
|
self.mmsport = mmsport
|
|
self.mmsproxy = mmsproxy
|
|
self.protocol = protocol
|
|
self.mvnoval = mvnoval
|
|
self.mvnotype = mvnotype
|
|
|
|
def set_apn(self):
|
|
apn_params = {
|
|
'carrier': self._apn_name,
|
|
'apn': self._apn,
|
|
'proxy': self.proxy or '',
|
|
'port': self.port or '',
|
|
'user': self.user or '',
|
|
'password': self.password or '',
|
|
'server': self.server or '',
|
|
'mmsc': self.mmsc or '',
|
|
'mmsport': self.mmsport or '',
|
|
'mmsproxy': self.mmsproxy or '',
|
|
'mcc': self._mcc,
|
|
'mnc': self._mnc,
|
|
'auth': self.auth or '-1',
|
|
'type': self.type or 'default',
|
|
'protocol': self.protocol or '',
|
|
'mvnotype': self.mvnotype or '',
|
|
'mvnoval': self.mvnoval or '',
|
|
}
|
|
self.dbg('APN parameters: ' + str(apn_params))
|
|
|
|
# search for carrier in database
|
|
carrier_id = self.get_carrier_id(apn_params['carrier'])
|
|
|
|
# add/update carrier
|
|
carrier_id = self.set_new_carrier(apn_params, carrier_id)
|
|
|
|
# select as preferred APN
|
|
if self.select:
|
|
self.set_preferred_apn(carrier_id)
|
|
|
|
def __str__(self):
|
|
return self.name()
|
|
|
|
def apn(self):
|
|
return self._apn
|
|
|
|
def mcc(self):
|
|
return self._mcc
|
|
|
|
def mnc(self):
|
|
return self._mnc
|
|
|
|
def select(self):
|
|
return self._select
|
|
|
|
# vim: expandtab tabstop=4 shiftwidth=4
|