Initial Open5GS support

Change-Id: Iff8b88dc22d8f156572839abb48b0c1377c55e33
This commit is contained in:
Pau Espin 2021-03-16 14:25:37 +01:00
parent b32ed9256f
commit 0696c60812
21 changed files with 2973 additions and 1 deletions

View File

@ -154,7 +154,7 @@ build_repo_dir() {
elif [ -f meson.build ]; then
rm -rf build && mkdir build && cd build || exit 1
set +x; echo; echo; set -x
meson ../ --prefix=$prefix $configure_opts
meson ../ --prefix=$prefix --libdir="lib" $configure_opts
ninja -j8
ninja install
return

View File

@ -392,6 +392,13 @@ pip3 install \
pydbus
----
If one plans to use Open5GS EPC, pymongo modules to interact against MongoDB
(_epc_open5gs.py_) shall be installed:
----
pip3 install \
pymongo
----
IMPORTANT: ofono may need to be installed from source to contain the most
recent fixes needed to operate your modems. This depends on the modem hardware
used and the tests run. Please see <<hardware_modems>>.

View File

@ -71,6 +71,9 @@ class EPC(log.Origin, metaclass=ABCMeta):
elif epc_type == 'srsepc':
from .epc_srs import srsEPC
epc_class = srsEPC
elif epc_type == 'open5gsepc':
from .epc_open5gs import Open5gsEPC
epc_class = Open5gsEPC
else:
raise log.Error('EPC type not supported:', epc_type)

View File

@ -0,0 +1,184 @@
# osmo_gsm_tester: specifics for running an Open5GS EPC
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
import copy
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
from .pcrf_open5gs import Open5gsPCRF
from .upf_open5gs import Open5gsUPF
from .smf_open5gs import Open5gsSMF
from .hss_open5gs import Open5gsHSS
from .mme_open5gs import Open5gsMME
from .sgwc_open5gs import Open5gsSGWC
from .sgwu_open5gs import Open5gsSGWU
def on_register_schemas():
config_schema = {
'db_host': schema.STR,
}
schema.register_config_schema('epc', config_schema)
class Open5gsEPC(epc.EPC):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
def __init__(self, testenv, run_node):
super().__init__(testenv, run_node, 'open5gs_epc')
self.run_dir = None
self.pcrf = None
self.upf = None
self.smf = None
self.mme = None
self.hss = None
self.sgwc = None
self.sgwu = None
self.subscriber_list = []
def cleanup(self):
if self.pcrf:
self.pcrf.cleanup()
if self.upf:
self.upf.cleanup()
if self.smf:
self.smf.cleanup()
if self.hss:
self.hss.cleanup()
if self.mme:
self.mme.cleanup()
if self.sgwc:
self.sgwc.cleanup()
if self.sgwu:
self.sgwu.cleanup()
def configure(self):
values = super().configure(['open5gsepc'])
db_host = values['epc']['db_host']
db_uri = 'mongodb://'+db_host+'/open5gs'
config.overlay(values, dict(epc=dict(db_uri=db_uri)))
self.fill_subscribers_mongodb(values['epc']['db_host'], 27017)
self.pcrf = Open5gsPCRF(self.testenv, self)
self.upf = Open5gsUPF(self.testenv, self)
self.smf = Open5gsSMF(self.testenv, self)
self.hss = Open5gsHSS(self.testenv, self)
self.mme = Open5gsMME(self.testenv, self)
self.sgwc = Open5gsSGWC(self.testenv, self)
self.sgwu = Open5gsSGWU(self.testenv, self)
self.pcrf.configure(copy.deepcopy(values))
self.upf.configure(copy.deepcopy(values))
self.smf.configure(copy.deepcopy(values))
self.hss.configure(copy.deepcopy(values))
self.mme.configure(copy.deepcopy(values))
self.sgwc.configure(copy.deepcopy(values))
self.sgwu.configure(copy.deepcopy(values))
def start(self):
self.log('Starting srsepc')
self.run_dir = util.Dir(self.testenv.test().get_run_dir().new_dir(self.name()))
self.configure()
self.pcrf.start()
self.upf.start()
self.smf.start()
self.hss.start()
self.mme.start()
self.sgwc.start()
self.sgwu.start()
def subscriber_add(self, modem, msisdn=None, algo_str=None):
if msisdn is None:
msisdn = modem.msisdn()
if algo_str is None:
algo_str = modem.auth_algo() or 'milenage'
if algo_str == 'milenage':
if not modem.ki():
raise log.Error("Auth algo milenage selected but no KI specified")
if not modem.opc():
raise log.Error("Auth algo milenage selected but no OPC specified")
else:
raise log.Error("Open5Gs only supports auth algo: milenage")
subscriber_id = len(self.subscriber_list) # list index
self.subscriber_list.append({'id': subscriber_id, 'imsi': modem.imsi(), 'msisdn': msisdn, 'auth_algo': algo_str, 'ki': modem.ki(), 'opc': modem.opc(), 'apn_ipaddr': modem.apn_ipaddr()})
return subscriber_id
def fill_subscribers_mongodb(self, server, port):
import pymongo
myclient = pymongo.MongoClient("mongodb://" + str(server) + ":" + str(port) + "/")
mydb = myclient["open5gs"]
mycol = mydb["subscribers"]
for s in self.subscriber_list:
self.log('Insert subscriber to DB', msisdn=s['msisdn'], imsi=s['imsi'], subscriber_id=s['id'],
algo_str=s['auth_algo'])
slice_data = [ { \
"sst": 1, \
"default_indicator": True, \
"session": [ \
{ \
"name": "internet", \
"type": 3, "pcc_rule": [], "ambr": {"uplink": {"value": 1, "unit": 0}, "downlink": {"value": 1, "unit": 0}}, \
"qos": { "index": 9, "arp": {"priority_level": 8, "pre_emption_capability": 1, "pre_emption_vulnerability": 1} } \
} \
] \
} ]
sub_data = {'imsi': s['imsi'], \
'subscribed_rau_tau_timer': 12, \
'network_access_mode': 2, \
'subscriber_status': 0, \
"access_restriction_data": 32, \
'slice': slice_data, \
'ambr': {"uplink": {"value": 1, "unit": 0}, "downlink": {"value": 1, "unit": 0}}, \
'security': {'k': s['ki'], 'amf': '8000', 'op': None, 'opc': s['opc']},
'schema_version': 1, \
'__v': 0}
x = mycol.insert_one(sub_data)
self.dbg("Added subscriber with Inserted ID : " + str(x.inserted_id))
s['inserted_id'] = x.inserted_id
def enb_is_connected(self, enb):
# Match against sample mmed line: "eNB-S1 accepted[172.18.50.101]:50867"
if not self.mme or not self.mme.running():
return False
stdout_lines = (self.mme.process.get_stdout() or '').splitlines()
for l in stdout_lines:
if 'eNB' in l and 'accepted' in l and enb.addr() in l:
return True
return False
def running(self):
return self.pcrf and self.upf and self.smf and self.hss and \
self.mme and self.sgwc and self.sgwu and \
self.pcrf.running() and self.upf.running() and self.smf.running() and \
self.hss.running() and self.mme.running() and self.sgwc.running() and \
self.sgwu.running()
def tun_addr(self):
return '172.16.0.1'
def get_kpis(self):
return {}
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,158 @@
# osmo_gsm_tester: specifics for running an Open5GS hssd process
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
def on_register_schemas():
pass
class Open5gsHSS(log.Origin):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
BINFILE = 'open5gs-hssd'
CFGFILE = 'open5gs-hssd.yaml'
LOGFILE = 'open5gs-hssd.log'
DIAMETERFILE = 'open5gs-freediameter.conf'
def __init__(self, testenv, o5gs_epc):
super().__init__(log.C_RUN, 'open5gs-hssd')
self.testenv = testenv
self.o5gs_epc = o5gs_epc
self._run_node = o5gs_epc.run_node()
self.run_dir = None
self.config_file = None
self.log_file = None
self.diameter_file = None
self.process = None
self.rem_host = None
self.remote_inst = None
self.remote_config_file = None
self.remote_log_file = None
self.remote_diameter_file = None
def cleanup(self):
if self.process is None:
return
if self._run_node.is_local():
return
# copy back files (may not exist, for instance if there was an early error of process):
try:
self.rem_host.scpfrom('scp-back-log', self.remote_log_file, self.log_file)
except Exception as e:
self.log(repr(e))
def start(self):
self.log('Starting %s' % Open5gsHSS.BINFILE)
if self._run_node.is_local():
self.start_locally()
else:
self.start_remotely()
def start_remotely(self):
remote_env = { 'LD_LIBRARY_PATH': self.remote_inst.child('lib') }
remote_lib = self.remote_inst.child('lib')
remote_binary = self.remote_inst.child('bin', Open5gsHSS.BINFILE)
args = (remote_binary, '-c', self.remote_config_file)
self.process = self.rem_host.RemoteProcess(Open5gsHSS.BINFILE, args, remote_env=remote_env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def start_locally(self):
binary = self.inst.child('bin', Open5gsHSS.BINFILE)
lib = self.inst.child('lib')
env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) }
args = (binary, '-c', os.path.abspath(self.config_file))
self.process = process.Process(self.name(), self.run_dir, args, env=env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def configure(self, values):
self.run_dir = util.Dir(self.o5gs_epc.run_dir.new_dir(self.name()))
self.inst = util.Dir(os.path.abspath(self.testenv.suite().trial().get_inst('open5gs', self._run_node.run_label())))
if not os.path.isdir(self.inst.child('lib')):
raise log.Error('No lib/ in', self.inst)
if not self.inst.isfile('bin', Open5gsHSS.BINFILE):
raise log.Error('No %s binary in' % Open5gsHSS.BINFILE, self.inst)
self.config_file = self.run_dir.child(Open5gsHSS.CFGFILE)
self.log_file = self.run_dir.child(Open5gsHSS.LOGFILE)
self.diameter_file = self.run_dir.child(Open5gsHSS.DIAMETERFILE)
if not self._run_node.is_local():
self.rem_host = remote.RemoteHost(self.run_dir, self._run_node.ssh_user(), self._run_node.ssh_addr())
remote_prefix_dir = util.Dir(Open5gsHSS.REMOTE_DIR)
self.remote_inst = util.Dir(remote_prefix_dir.child(os.path.basename(str(self.inst))))
remote_run_dir = util.Dir(remote_prefix_dir.child(Open5gsHSS.BINFILE))
self.remote_config_file = remote_run_dir.child(Open5gsHSS.CFGFILE)
self.remote_log_file = remote_run_dir.child(Open5gsHSS.LOGFILE)
self.remote_diameter_file = remote_run_dir.child(Open5gsHSS.DIAMETERFILE)
logfile = self.log_file if self._run_node.is_local() else self.remote_log_file
diameter_file = self.diameter_file if self._run_node.is_local() else self.remote_diameter_file
inst_prefix = str(self.inst) if self._run_node.is_local() else str(self.remote_inst)
config.overlay(values, dict(hss=dict(log_filename=logfile,
diameter_filename=diameter_file,
inst_prefix=inst_prefix)))
config.overlay(values, dict(diameter=dict(identity=self.diameter_name(),
inst_prefix=inst_prefix,
listen_address=self.o5gs_epc.addr(),
listen_port=self.diameter_port(),
connect_name=self.o5gs_epc.mme.diameter_name(),
connect_address=self.o5gs_epc.addr(),
connect_port=self.o5gs_epc.mme.diameter_port())))
self.dbg('OPEN5GS-HSS CONFIG:\n' + pprint.pformat(values))
with open(self.config_file, 'w') as f:
r = template.render(Open5gsHSS.CFGFILE, values)
self.dbg(r)
f.write(r)
with open(self.diameter_file, 'w') as f:
r = template.render(Open5gsHSS.DIAMETERFILE, values)
self.dbg(r)
f.write(r)
if not self._run_node.is_local():
self.rem_host.recreate_remote_dir(self.remote_inst)
self.rem_host.scp('scp-inst-to-remote', str(self.inst), remote_prefix_dir)
self.rem_host.recreate_remote_dir(remote_run_dir)
self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
self.rem_host.scp('scp-diam-to-remote', self.diameter_file, self.remote_diameter_file)
def running(self):
return not self.process.terminated()
def diameter_name(self):
return 'hss'
def diameter_port(self):
return 3868;
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,172 @@
# osmo_gsm_tester: specifics for running an Open5GS mmed process
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
def on_register_schemas():
pass
class Open5gsMME(log.Origin):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
BINFILE = 'open5gs-mmed'
CFGFILE = 'open5gs-mmed.yaml'
LOGFILE = 'open5gs-mmed.log'
DIAMETERFILE = 'open5gs-freediameter.conf'
def __init__(self, testenv, o5gs_epc):
super().__init__(log.C_RUN, 'open5gs-mmed')
self.testenv = testenv
self.o5gs_epc = o5gs_epc
self._run_node = o5gs_epc.run_node()
self.run_dir = None
self.config_file = None
self.log_file = None
self.diameter_file = None
self.process = None
self.rem_host = None
self.remote_inst = None
self.remote_config_file = None
self.remote_log_file = None
self.remote_diameter_file = None
def cleanup(self):
if self.process is None:
return
if self._run_node.is_local():
return
# copy back files (may not exist, for instance if there was an early error of process):
try:
self.rem_host.scpfrom('scp-back-log', self.remote_log_file, self.log_file)
except Exception as e:
self.log(repr(e))
def start(self):
self.log('Starting %s' % Open5gsMME.BINFILE)
if self._run_node.is_local():
self.start_locally()
else:
self.start_remotely()
def start_remotely(self):
remote_lib = self.remote_inst.child('lib')
remote_binary = self.remote_inst.child('bin', Open5gsMME.BINFILE)
# setting capabilities will later disable use of LD_LIBRARY_PATH from ELF loader -> modify RPATH instead.
self.log('Setting RPATH for open5gs-mmed')
self.rem_host.change_elf_rpath(remote_binary, remote_lib)
# open5gs-mmed requires CAP_NET_ADMIN to create tunnel devices: ioctl(TUNSETIFF):
self.log('Applying CAP_NET_ADMIN capability to open5gs-mmed')
self.rem_host.setcap_net_admin(remote_binary)
args = (remote_binary, '-c', self.remote_config_file)
self.process = self.rem_host.RemoteProcess(Open5gsMME.BINFILE, args)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def start_locally(self):
binary = self.inst.child('bin', Open5gsMME.BINFILE)
lib = self.inst.child('lib')
env = {}
# setting capabilities will later disable use of LD_LIBRARY_PATH from ELF loader -> modify RPATH instead.
self.log('Setting RPATH for open5gs-mmed')
# open5gs-mmed binary needs patchelf <= 0.9 (0.10 and current master fail) to avoid failing during patch. OS#4389, patchelf-GH#192.
util.change_elf_rpath(binary, util.prepend_library_path(lib), self.run_dir.new_dir('patchelf'))
# open5gs-mmed requires CAP_NET_ADMIN to create tunnel devices: ioctl(TUNSETIFF):
self.log('Applying CAP_NET_ADMIN capability to open5gs-mmed')
util.setcap_net_admin(binary, self.run_dir.new_dir('setcap_net_admin'))
args = (binary, '-c', os.path.abspath(self.config_file))
self.process = process.Process(self.name(), self.run_dir, args, env=env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def configure(self, values):
self.run_dir = util.Dir(self.o5gs_epc.run_dir.new_dir(self.name()))
self.inst = util.Dir(os.path.abspath(self.testenv.suite().trial().get_inst('open5gs', self._run_node.run_label())))
if not os.path.isdir(self.inst.child('lib')):
raise log.Error('No lib/ in', self.inst)
if not self.inst.isfile('bin', Open5gsMME.BINFILE):
raise log.Error('No %s binary in' % Open5gsMME.BINFILE, self.inst)
self.config_file = self.run_dir.child(Open5gsMME.CFGFILE)
self.log_file = self.run_dir.child(Open5gsMME.LOGFILE)
self.diameter_file = self.run_dir.child(Open5gsMME.DIAMETERFILE)
if not self._run_node.is_local():
self.rem_host = remote.RemoteHost(self.run_dir, self._run_node.ssh_user(), self._run_node.ssh_addr())
remote_prefix_dir = util.Dir(Open5gsMME.REMOTE_DIR)
self.remote_inst = util.Dir(remote_prefix_dir.child(os.path.basename(str(self.inst))))
remote_run_dir = util.Dir(remote_prefix_dir.child(Open5gsMME.BINFILE))
self.remote_config_file = remote_run_dir.child(Open5gsMME.CFGFILE)
self.remote_log_file = remote_run_dir.child(Open5gsMME.LOGFILE)
self.remote_diameter_file = remote_run_dir.child(Open5gsMME.DIAMETERFILE)
logfile = self.log_file if self._run_node.is_local() else self.remote_log_file
diameter_file = self.diameter_file if self._run_node.is_local() else self.remote_diameter_file
inst_prefix = str(self.inst) if self._run_node.is_local() else str(self.remote_inst)
config.overlay(values, dict(mme=dict(log_filename=logfile,
diameter_filename=diameter_file,
inst_prefix=inst_prefix)))
config.overlay(values, dict(diameter=dict(identity=self.diameter_name(),
inst_prefix=inst_prefix,
listen_address=self.o5gs_epc.addr(),
listen_port=self.diameter_port(),
connect_name=self.o5gs_epc.hss.diameter_name(),
connect_address=self.o5gs_epc.addr(),
connect_port=self.o5gs_epc.hss.diameter_port())))
self.dbg('OPEN5GS-MME CONFIG:\n' + pprint.pformat(values))
with open(self.config_file, 'w') as f:
r = template.render(Open5gsMME.CFGFILE, values)
self.dbg(r)
f.write(r)
with open(self.diameter_file, 'w') as f:
r = template.render(Open5gsMME.DIAMETERFILE, values)
self.dbg(r)
f.write(r)
if not self._run_node.is_local():
self.rem_host.recreate_remote_dir(self.remote_inst)
self.rem_host.scp('scp-inst-to-remote', str(self.inst), remote_prefix_dir)
self.rem_host.recreate_remote_dir(remote_run_dir)
self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
self.rem_host.scp('scp-diam-to-remote', self.diameter_file, self.remote_diameter_file)
def running(self):
return not self.process.terminated()
def diameter_name(self):
return 'mme'
def diameter_port(self):
return 3868 + 1;
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,158 @@
# osmo_gsm_tester: specifics for running an Open5GS pcrfd process
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
def on_register_schemas():
pass
class Open5gsPCRF(log.Origin):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
BINFILE = 'open5gs-pcrfd'
CFGFILE = 'open5gs-pcrfd.yaml'
LOGFILE = 'open5gs-pcrfd.log'
DIAMETERFILE = 'open5gs-freediameter.conf'
def __init__(self, testenv, o5gs_epc):
super().__init__(log.C_RUN, 'open5gs-pcrfd')
self.testenv = testenv
self.o5gs_epc = o5gs_epc
self._run_node = o5gs_epc.run_node()
self.run_dir = None
self.config_file = None
self.log_file = None
self.diameter_file = None
self.process = None
self.rem_host = None
self.remote_inst = None
self.remote_config_file = None
self.remote_log_file = None
self.remote_diameter_file = None
def cleanup(self):
if self.process is None:
return
if self._run_node.is_local():
return
# copy back files (may not exist, for instance if there was an early error of process):
try:
self.rem_host.scpfrom('scp-back-log', self.remote_log_file, self.log_file)
except Exception as e:
self.log(repr(e))
def start(self):
self.log('Starting %s' % Open5gsPCRF.BINFILE)
if self._run_node.is_local():
self.start_locally()
else:
self.start_remotely()
def start_remotely(self):
remote_env = { 'LD_LIBRARY_PATH': self.remote_inst.child('lib') }
remote_lib = self.remote_inst.child('lib')
remote_binary = self.remote_inst.child('bin', Open5gsPCRF.BINFILE)
args = (remote_binary, '-c', self.remote_config_file)
self.process = self.rem_host.RemoteProcess(Open5gsPCRF.BINFILE, args, remote_env=remote_env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def start_locally(self):
binary = self.inst.child('bin', Open5gsPCRF.BINFILE)
lib = self.inst.child('lib')
env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) }
args = (binary, '-c', os.path.abspath(self.config_file))
self.process = process.Process(self.name(), self.run_dir, args, env=env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def configure(self, values):
self.run_dir = util.Dir(self.o5gs_epc.run_dir.new_dir(self.name()))
self.inst = util.Dir(os.path.abspath(self.testenv.suite().trial().get_inst('open5gs', self._run_node.run_label())))
if not os.path.isdir(self.inst.child('lib')):
raise log.Error('No lib/ in', self.inst)
if not self.inst.isfile('bin', Open5gsPCRF.BINFILE):
raise log.Error('No %s binary in' % Open5gsPCRF.BINFILE, self.inst)
self.config_file = self.run_dir.child(Open5gsPCRF.CFGFILE)
self.log_file = self.run_dir.child(Open5gsPCRF.LOGFILE)
self.diameter_file = self.run_dir.child(Open5gsPCRF.DIAMETERFILE)
if not self._run_node.is_local():
self.rem_host = remote.RemoteHost(self.run_dir, self._run_node.ssh_user(), self._run_node.ssh_addr())
remote_prefix_dir = util.Dir(Open5gsPCRF.REMOTE_DIR)
self.remote_inst = util.Dir(remote_prefix_dir.child(os.path.basename(str(self.inst))))
remote_run_dir = util.Dir(remote_prefix_dir.child(Open5gsPCRF.BINFILE))
self.remote_config_file = remote_run_dir.child(Open5gsPCRF.CFGFILE)
self.remote_log_file = remote_run_dir.child(Open5gsPCRF.LOGFILE)
self.remote_diameter_file = remote_run_dir.child(Open5gsPCRF.DIAMETERFILE)
logfile = self.log_file if self._run_node.is_local() else self.remote_log_file
diameter_file = self.diameter_file if self._run_node.is_local() else self.remote_diameter_file
inst_prefix = str(self.inst) if self._run_node.is_local() else str(self.remote_inst)
config.overlay(values, dict(pcrf=dict(log_filename=logfile,
diameter_filename=diameter_file,
inst_prefix=inst_prefix)))
config.overlay(values, dict(diameter=dict(identity=self.diameter_name(),
inst_prefix=inst_prefix,
listen_address=self.o5gs_epc.addr(),
listen_port=self.diameter_port(),
connect_name=self.o5gs_epc.smf.diameter_name(),
connect_address=self.o5gs_epc.addr(),
connect_port=self.o5gs_epc.smf.diameter_port())))
self.dbg('OPEN5GS-PCRF CONFIG:\n' + pprint.pformat(values))
with open(self.config_file, 'w') as f:
r = template.render(Open5gsPCRF.CFGFILE, values)
self.dbg(r)
f.write(r)
with open(self.diameter_file, 'w') as f:
r = template.render(Open5gsPCRF.DIAMETERFILE, values)
self.dbg(r)
f.write(r)
if not self._run_node.is_local():
self.rem_host.recreate_remote_dir(self.remote_inst)
self.rem_host.scp('scp-inst-to-remote', str(self.inst), remote_prefix_dir)
self.rem_host.recreate_remote_dir(remote_run_dir)
self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
self.rem_host.scp('scp-diam-to-remote', self.diameter_file, self.remote_diameter_file)
def running(self):
return not self.process.terminated()
def diameter_name(self):
return 'pcrf'
def diameter_port(self):
return 3868 + 2;
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,132 @@
# osmo_gsm_tester: specifics for running an Open5GS swgcd process
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
def on_register_schemas():
pass
class Open5gsSGWC(log.Origin):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
BINFILE = 'open5gs-sgwcd'
CFGFILE = 'open5gs-sgwcd.yaml'
LOGFILE = 'open5gs-sgwcd.log'
def __init__(self, testenv, o5gs_epc):
super().__init__(log.C_RUN, 'open5gs-sgwcd')
self.testenv = testenv
self.o5gs_epc = o5gs_epc
self._run_node = o5gs_epc.run_node()
self.run_dir = None
self.config_file = None
self.log_file = None
self.process = None
self.rem_host = None
self.remote_inst = None
self.remote_config_file = None
self.remote_log_file = None
def cleanup(self):
if self.process is None:
return
if self._run_node.is_local():
return
# copy back files (may not exist, for instance if there was an early error of process):
try:
self.rem_host.scpfrom('scp-back-log', self.remote_log_file, self.log_file)
except Exception as e:
self.log(repr(e))
def start(self):
self.log('Starting %s' % Open5gsSGWC.BINFILE)
if self._run_node.is_local():
self.start_locally()
else:
self.start_remotely()
def start_remotely(self):
remote_env = { 'LD_LIBRARY_PATH': self.remote_inst.child('lib') }
remote_lib = self.remote_inst.child('lib')
remote_binary = self.remote_inst.child('bin', Open5gsSGWC.BINFILE)
args = (remote_binary, '-c', self.remote_config_file)
self.process = self.rem_host.RemoteProcess(Open5gsSGWC.BINFILE, args, remote_env=remote_env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def start_locally(self):
binary = self.inst.child('bin', Open5gsSGWC.BINFILE)
lib = self.inst.child('lib')
env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) }
args = (binary, '-c', os.path.abspath(self.config_file))
self.process = process.Process(self.name(), self.run_dir, args, env=env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def configure(self, values):
self.run_dir = util.Dir(self.o5gs_epc.run_dir.new_dir(self.name()))
self.inst = util.Dir(os.path.abspath(self.testenv.suite().trial().get_inst('open5gs', self._run_node.run_label())))
if not os.path.isdir(self.inst.child('lib')):
raise log.Error('No lib/ in', self.inst)
if not self.inst.isfile('bin', Open5gsSGWC.BINFILE):
raise log.Error('No %s binary in' % Open5gsSGWC.BINFILE, self.inst)
self.config_file = self.run_dir.child(Open5gsSGWC.CFGFILE)
self.log_file = self.run_dir.child(Open5gsSGWC.LOGFILE)
if not self._run_node.is_local():
self.rem_host = remote.RemoteHost(self.run_dir, self._run_node.ssh_user(), self._run_node.ssh_addr())
remote_prefix_dir = util.Dir(Open5gsSGWC.REMOTE_DIR)
self.remote_inst = util.Dir(remote_prefix_dir.child(os.path.basename(str(self.inst))))
remote_run_dir = util.Dir(remote_prefix_dir.child(Open5gsSGWC.BINFILE))
self.remote_config_file = remote_run_dir.child(Open5gsSGWC.CFGFILE)
self.remote_log_file = remote_run_dir.child(Open5gsSGWC.LOGFILE)
logfile = self.log_file if self._run_node.is_local() else self.remote_log_file
inst_prefix = str(self.inst) if self._run_node.is_local() else str(self.remote_inst)
config.overlay(values, dict(sgwc=dict(log_filename=logfile,
inst_prefix=inst_prefix)))
self.dbg('OPEN5GS-SGWC CONFIG:\n' + pprint.pformat(values))
with open(self.config_file, 'w') as f:
r = template.render(Open5gsSGWC.CFGFILE, values)
self.dbg(r)
f.write(r)
if not self._run_node.is_local():
self.rem_host.recreate_remote_dir(self.remote_inst)
self.rem_host.scp('scp-inst-to-remote', str(self.inst), remote_prefix_dir)
self.rem_host.recreate_remote_dir(remote_run_dir)
self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
def running(self):
return not self.process.terminated()
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,140 @@
# osmo_gsm_tester: specifics for running an Open5GS sgwud process
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
def on_register_schemas():
pass
class Open5gsSGWU(log.Origin):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
BINFILE = 'open5gs-sgwud'
CFGFILE = 'open5gs-sgwud.yaml'
LOGFILE = 'open5gs-sgwud.log'
def __init__(self, testenv, o5gs_epc):
super().__init__(log.C_RUN, 'open5gs-sgwud')
self.testenv = testenv
self.o5gs_epc = o5gs_epc
self._run_node = o5gs_epc.run_node()
self.run_dir = None
self.config_file = None
self.log_file = None
self.process = None
self.rem_host = None
self.remote_inst = None
self.remote_config_file = None
self.remote_log_file = None
def cleanup(self):
if self.process is None:
return
if self._run_node.is_local():
return
# copy back files (may not exist, for instance if there was an early error of process):
try:
self.rem_host.scpfrom('scp-back-log', self.remote_log_file, self.log_file)
except Exception as e:
self.log(repr(e))
def start(self):
self.log('Starting %s' % Open5gsSGWU.BINFILE)
if self._run_node.is_local():
self.start_locally()
else:
self.start_remotely()
def start_remotely(self):
remote_env = { 'LD_LIBRARY_PATH': self.remote_inst.child('lib') }
remote_lib = self.remote_inst.child('lib')
remote_binary = self.remote_inst.child('bin', Open5gsSGWU.BINFILE)
args = (remote_binary, '-c', self.remote_config_file)
self.process = self.rem_host.RemoteProcess(Open5gsSGWU.BINFILE, args, remote_env=remote_env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def start_locally(self):
binary = self.inst.child('bin', Open5gsSGWU.BINFILE)
lib = self.inst.child('lib')
env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) }
# setting capabilities will later disable use of LD_LIBRARY_PATH from ELF loader -> modify RPATH instead.
self.log('Setting RPATH for open5gs-sgwud')
# open5gs-sgwud binary needs patchelf <= 0.9 (0.10 and current master fail) to avoid failing during patch. OS#4389, patchelf-GH#192.
util.change_elf_rpath(binary, util.prepend_library_path(lib), self.run_dir.new_dir('patchelf'))
# open5gs-sgwud requires CAP_NET_ADMIN to create tunnel devices: ioctl(TUNSETIFF):
self.log('Applying CAP_NET_ADMIN capability to open5gs-sgwud')
util.setcap_net_admin(binary, self.run_dir.new_dir('setcap_net_admin'))
args = (binary, '-c', os.path.abspath(self.config_file))
self.process = process.Process(self.name(), self.run_dir, args, env=env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def configure(self, values):
self.run_dir = util.Dir(self.o5gs_epc.run_dir.new_dir(self.name()))
self.inst = util.Dir(os.path.abspath(self.testenv.suite().trial().get_inst('open5gs', self._run_node.run_label())))
if not os.path.isdir(self.inst.child('lib')):
raise log.Error('No lib/ in', self.inst)
if not self.inst.isfile('bin', Open5gsSGWU.BINFILE):
raise log.Error('No %s binary in' % Open5gsSGWU.BINFILE, self.inst)
self.config_file = self.run_dir.child(Open5gsSGWU.CFGFILE)
self.log_file = self.run_dir.child(Open5gsSGWU.LOGFILE)
if not self._run_node.is_local():
self.rem_host = remote.RemoteHost(self.run_dir, self._run_node.ssh_user(), self._run_node.ssh_addr())
remote_prefix_dir = util.Dir(Open5gsSGWU.REMOTE_DIR)
self.remote_inst = util.Dir(remote_prefix_dir.child(os.path.basename(str(self.inst))))
remote_run_dir = util.Dir(remote_prefix_dir.child(Open5gsSGWU.BINFILE))
self.remote_config_file = remote_run_dir.child(Open5gsSGWU.CFGFILE)
self.remote_log_file = remote_run_dir.child(Open5gsSGWU.LOGFILE)
logfile = self.log_file if self._run_node.is_local() else self.remote_log_file
inst_prefix = str(self.inst) if self._run_node.is_local() else str(self.remote_inst)
config.overlay(values, dict(sgwu=dict(log_filename=logfile,
inst_prefix=inst_prefix)))
self.dbg('OPEN5GS-SGWU CONFIG:\n' + pprint.pformat(values))
with open(self.config_file, 'w') as f:
r = template.render(Open5gsSGWU.CFGFILE, values)
self.dbg(r)
f.write(r)
if not self._run_node.is_local():
self.rem_host.recreate_remote_dir(self.remote_inst)
self.rem_host.scp('scp-inst-to-remote', str(self.inst), remote_prefix_dir)
self.rem_host.recreate_remote_dir(remote_run_dir)
self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
def running(self):
return not self.process.terminated()
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,158 @@
# osmo_gsm_tester: specifics for running an Open5GS smfd process
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
def on_register_schemas():
pass
class Open5gsSMF(log.Origin):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
BINFILE = 'open5gs-smfd'
CFGFILE = 'open5gs-smfd.yaml'
LOGFILE = 'open5gs-smfd.log'
DIAMETERFILE = 'open5gs-freediameter.conf'
def __init__(self, testenv, o5gs_epc):
super().__init__(log.C_RUN, 'open5gs-smfd')
self.testenv = testenv
self.o5gs_epc = o5gs_epc
self._run_node = o5gs_epc.run_node()
self.run_dir = None
self.config_file = None
self.log_file = None
self.diameter_file = None
self.process = None
self.rem_host = None
self.remote_inst = None
self.remote_config_file = None
self.remote_log_file = None
self.remote_diameter_file = None
def cleanup(self):
if self.process is None:
return
if self._run_node.is_local():
return
# copy back files (may not exist, for instance if there was an early error of process):
try:
self.rem_host.scpfrom('scp-back-log', self.remote_log_file, self.log_file)
except Exception as e:
self.log(repr(e))
def start(self):
self.log('Starting %s' % Open5gsSMF.BINFILE)
if self._run_node.is_local():
self.start_locally()
else:
self.start_remotely()
def start_remotely(self):
remote_env = { 'LD_LIBRARY_PATH': self.remote_inst.child('lib') }
remote_lib = self.remote_inst.child('lib')
remote_binary = self.remote_inst.child('bin', Open5gsSMF.BINFILE)
args = (remote_binary, '-c', self.remote_config_file)
self.process = self.rem_host.RemoteProcess(Open5gsSMF.BINFILE, args, remote_env=remote_env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def start_locally(self):
binary = self.inst.child('bin', Open5gsSMF.BINFILE)
lib = self.inst.child('lib')
env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) }
args = (binary, '-c', os.path.abspath(self.config_file))
self.process = process.Process(self.name(), self.run_dir, args, env=env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def configure(self, values):
self.run_dir = util.Dir(self.o5gs_epc.run_dir.new_dir(self.name()))
self.inst = util.Dir(os.path.abspath(self.testenv.suite().trial().get_inst('open5gs', self._run_node.run_label())))
if not os.path.isdir(self.inst.child('lib')):
raise log.Error('No lib/ in', self.inst)
if not self.inst.isfile('bin', Open5gsSMF.BINFILE):
raise log.Error('No %s binary in' % Open5gsSMF.BINFILE, self.inst)
self.config_file = self.run_dir.child(Open5gsSMF.CFGFILE)
self.log_file = self.run_dir.child(Open5gsSMF.LOGFILE)
self.diameter_file = self.run_dir.child(Open5gsSMF.DIAMETERFILE)
if not self._run_node.is_local():
self.rem_host = remote.RemoteHost(self.run_dir, self._run_node.ssh_user(), self._run_node.ssh_addr())
remote_prefix_dir = util.Dir(Open5gsSMF.REMOTE_DIR)
self.remote_inst = util.Dir(remote_prefix_dir.child(os.path.basename(str(self.inst))))
remote_run_dir = util.Dir(remote_prefix_dir.child(Open5gsSMF.BINFILE))
self.remote_config_file = remote_run_dir.child(Open5gsSMF.CFGFILE)
self.remote_log_file = remote_run_dir.child(Open5gsSMF.LOGFILE)
self.remote_diameter_file = remote_run_dir.child(Open5gsSMF.DIAMETERFILE)
logfile = self.log_file if self._run_node.is_local() else self.remote_log_file
diameter_file = self.diameter_file if self._run_node.is_local() else self.remote_diameter_file
inst_prefix = str(self.inst) if self._run_node.is_local() else str(self.remote_inst)
config.overlay(values, dict(smf=dict(log_filename=logfile,
diameter_filename=diameter_file,
inst_prefix=inst_prefix)))
config.overlay(values, dict(diameter=dict(identity=self.diameter_name(),
inst_prefix=inst_prefix,
listen_address=self.o5gs_epc.addr(),
listen_port=self.diameter_port(),
connect_name=self.o5gs_epc.pcrf.diameter_name(),
connect_address=self.o5gs_epc.addr(),
connect_port=self.o5gs_epc.pcrf.diameter_port())))
self.dbg('OPEN5GS-SMF CONFIG:\n' + pprint.pformat(values))
with open(self.config_file, 'w') as f:
r = template.render(Open5gsSMF.CFGFILE, values)
self.dbg(r)
f.write(r)
with open(self.diameter_file, 'w') as f:
r = template.render(Open5gsSMF.DIAMETERFILE, values)
self.dbg(r)
f.write(r)
if not self._run_node.is_local():
self.rem_host.recreate_remote_dir(self.remote_inst)
self.rem_host.scp('scp-inst-to-remote', str(self.inst), remote_prefix_dir)
self.rem_host.recreate_remote_dir(remote_run_dir)
self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
self.rem_host.scp('scp-diam-to-remote', self.diameter_file, self.remote_diameter_file)
def running(self):
return not self.process.terminated()
def diameter_name(self):
return 'smf'
def diameter_port(self):
return 3868 + 3;
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,147 @@
# osmo_gsm_tester: specifics for running an Open%GS upfd process
#
# Copyright (C) 2021 by sysmocom - s.f.m.c. GmbH
#
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
#
# 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 os
import pprint
from ..core import log, util, config, template, process, remote
from ..core import schema
from . import epc
def on_register_schemas():
pass
class Open5gsUPF(log.Origin):
REMOTE_DIR = '/osmo-gsm-tester-open5gs'
BINFILE = 'open5gs-upfd'
CFGFILE = 'open5gs-upfd.yaml'
LOGFILE = 'open5gs-upfd.log'
DIAMETERFILE = 'open5gs-freediameter.conf'
def __init__(self, testenv, o5gs_epc):
super().__init__(log.C_RUN, 'open5gs-upfd')
self.testenv = testenv
self.o5gs_epc = o5gs_epc
self._run_node = o5gs_epc.run_node()
self.run_dir = None
self.config_file = None
self.log_file = None
self.process = None
self.rem_host = None
self.remote_inst = None
self.remote_config_file = None
self.remote_log_file = None
def cleanup(self):
if self.process is None:
return
if self._run_node.is_local():
return
# copy back files (may not exist, for instance if there was an early error of process):
try:
self.rem_host.scpfrom('scp-back-log', self.remote_log_file, self.log_file)
except Exception as e:
self.log(repr(e))
def start(self):
self.log('Starting %s' % Open5gsUPF.BINFILE)
if self._run_node.is_local():
self.start_locally()
else:
self.start_remotely()
def start_remotely(self):
remote_lib = self.remote_inst.child('lib')
remote_binary = self.remote_inst.child('bin', Open5gsUPF.BINFILE)
# setting capabilities will later disable use of LD_LIBRARY_PATH from ELF loader -> modify RPATH instead.
self.log('Setting RPATH for open5gs-upfd')
self.rem_host.change_elf_rpath(remote_binary, remote_lib)
# open5gs-upfd requires CAP_NET_ADMIN to create tunnel devices: ioctl(TUNSETIFF):
self.log('Applying CAP_NET_ADMIN capability to open5gs-upfd')
self.rem_host.setcap_net_admin(remote_binary)
args = (remote_binary, '-c', self.remote_config_file)
self.process = self.rem_host.RemoteProcess(Open5gsUPF.BINFILE, args)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def start_locally(self):
binary = self.inst.child('bin', Open5gsUPF.BINFILE)
lib = self.inst.child('lib')
env = {}
# setting capabilities will later disable use of LD_LIBRARY_PATH from ELF loader -> modify RPATH instead.
self.log('Setting RPATH for open5gs-upfd')
# open5gs-upfd binary needs patchelf <= 0.9 (0.10 and current master fail) to avoid failing during patch. OS#4389, patchelf-GH#192.
util.change_elf_rpath(binary, util.prepend_library_path(lib), self.run_dir.new_dir('patchelf'))
# open5gs-upfd requires CAP_NET_ADMIN to create tunnel devices: ioctl(TUNSETIFF):
self.log('Applying CAP_NET_ADMIN capability to open5gs-upfd')
util.setcap_net_admin(binary, self.run_dir.new_dir('setcap_net_admin'))
args = (binary, '-c', os.path.abspath(self.config_file))
self.process = process.Process(self.name(), self.run_dir, args, env=env)
self.testenv.remember_to_stop(self.process)
self.process.launch()
def configure(self, values):
self.run_dir = util.Dir(self.o5gs_epc.run_dir.new_dir(self.name()))
self.inst = util.Dir(os.path.abspath(self.testenv.suite().trial().get_inst('open5gs', self._run_node.run_label())))
if not os.path.isdir(self.inst.child('lib')):
raise log.Error('No lib/ in', self.inst)
if not self.inst.isfile('bin', Open5gsUPF.BINFILE):
raise log.Error('No %s binary in' % Open5gsUPF.BINFILE, self.inst)
self.config_file = self.run_dir.child(Open5gsUPF.CFGFILE)
self.log_file = self.run_dir.child(Open5gsUPF.LOGFILE)
if not self._run_node.is_local():
self.rem_host = remote.RemoteHost(self.run_dir, self._run_node.ssh_user(), self._run_node.ssh_addr())
remote_prefix_dir = util.Dir(Open5gsUPF.REMOTE_DIR)
self.remote_inst = util.Dir(remote_prefix_dir.child(os.path.basename(str(self.inst))))
remote_run_dir = util.Dir(remote_prefix_dir.child(Open5gsUPF.BINFILE))
self.remote_config_file = remote_run_dir.child(Open5gsUPF.CFGFILE)
self.remote_log_file = remote_run_dir.child(Open5gsUPF.LOGFILE)
logfile = self.log_file if self._run_node.is_local() else self.remote_log_file
inst_prefix = str(self.inst) if self._run_node.is_local() else str(self.remote_inst)
config.overlay(values, dict(upf=dict(log_filename=logfile,
inst_prefix=inst_prefix)))
self.dbg('OPEN5GS-UPF CONFIG:\n' + pprint.pformat(values))
with open(self.config_file, 'w') as f:
r = template.render(Open5gsUPF.CFGFILE, values)
self.dbg(r)
f.write(r)
if not self._run_node.is_local():
self.rem_host.recreate_remote_dir(self.remote_inst)
self.rem_host.scp('scp-inst-to-remote', str(self.inst), remote_prefix_dir)
self.rem_host.recreate_remote_dir(remote_run_dir)
self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
def running(self):
return not self.process.terminated()
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -0,0 +1,266 @@
# This is a sample configuration file for freeDiameter daemon.
# Most of the options can be omitted, as they default to reasonable values.
# Only TLS-related options must be configured properly in usual setups.
# It is possible to use "include" keyword to import additional files
# e.g.: include "/etc/freeDiameter.d/*.conf"
# This is exactly equivalent as copy & paste the content of the included file(s)
# where the "include" keyword is found.
##############################################################
## Peer identity and realm
# The Diameter Identity of this daemon.
# This must be a valid FQDN that resolves to the local host.
# Default: hostname's FQDN
#Identity = "aaa.koganei.freediameter.net";
Identity = "${diameter.identity}.localdomain";
# The Diameter Realm of this daemon.
# Default: the domain part of Identity (after the first dot).
#Realm = "koganei.freediameter.net";
Realm = "localdomain";
##############################################################
## Transport protocol configuration
# The port this peer is listening on for incoming connections (TCP and SCTP).
# Default: 3868. Use 0 to disable.
Port = ${diameter.listen_port};
# The port this peer is listening on for incoming TLS-protected connections (TCP and SCTP).
# See TLS_old_method for more information about TLS flavours.
# Note: we use TLS/SCTP instead of DTLS/SCTP at the moment. This will change in future version of freeDiameter.
# Default: 5868. Use 0 to disable.
SecPort = 0;
# Use RFC3588 method for TLS protection, where TLS is negociated after CER/CEA exchange is completed
# on the unsecure connection. The alternative is RFC6733 mechanism, where TLS protects also the
# CER/CEA exchange on a dedicated secure port.
# This parameter only affects outgoing connections.
# The setting can be also defined per-peer (see Peers configuration section).
# Default: use RFC6733 method with separate port for TLS.
#TLS_old_method;
# Disable use of TCP protocol (only listen and connect over SCTP)
# Default : TCP enabled
#No_TCP;
# Disable use of SCTP protocol (only listen and connect over TCP)
# Default : SCTP enabled
#No_SCTP;
# This option is ignored if freeDiameter is compiled with DISABLE_SCTP option.
# Prefer TCP instead of SCTP for establishing new connections.
# This setting may be overwritten per peer in peer configuration blocs.
# Default : SCTP is attempted first.
#Prefer_TCP;
# Default number of streams per SCTP associations.
# This setting may be overwritten per peer basis.
# Default : 30 streams
#SCTP_streams = 30;
##############################################################
## Endpoint configuration
# Disable use of IP addresses (only IPv6)
# Default : IP enabled
#No_IP;
# Disable use of IPv6 addresses (only IP)
# Default : IPv6 enabled
#No_IPv6;
# Specify local addresses the server must bind to
# Default : listen on all addresses available.
#ListenOn = "202.249.37.5";
#ListenOn = "2001:200:903:2::202:1";
#ListenOn = "fe80::21c:5ff:fe98:7d62%eth0";
ListenOn = "${diameter.listen_address}";