#!/usr/bin/env python3 # Copyright 2020 Canonical 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 sys sys.path.append("lib") from ops.charm import CharmBase from ops.framework import StoredState from ops.main import main from ops.model import ( ActiveStatus, BlockedStatus, MaintenanceStatus, WaitingStatus, ModelError, ) import subprocess import logging import asyncio import random logger = logging.getLogger(__name__) class EnodebCharm(CharmBase): state = StoredState() def __init__(self, *args): super().__init__(*args) # An example of setting charm state # that's persistent across events self.state.set_default(is_started=False) if not self.state.is_started: self.state.is_started = True # Register all of the events we want to observe for event in ( # Charm events self.on.config_changed, self.on.install, self.on.upgrade_charm, self.on.register_action, self.on.attach_ue_action, self.on.unregister_action, self.on.detach_ue_action, self.on.remove_default_gw_action, ): self.framework.observe(event, self) def on_config_changed(self, event): """Handle changes in configuration""" unit = self.model.unit def on_install(self, event): """Called when the charm is being installed""" unit = self.model.unit # Install your software and its dependencies unit.status = ActiveStatus() def on_upgrade_charm(self, event): """Upgrade the charm.""" unit = self.model.unit # Mark the unit as under Maintenance. unit.status = MaintenanceStatus("Upgrading charm") self.on_install(event) # When maintenance is done, return to an Active state unit.status = ActiveStatus() def on_register_action(self, event): """Register to AGW (EPC).""" try: mme_addr = event.params["mme-addr"] gtp_bind_addr = event.params["gtp-bind-addr"] s1c_bind_addr = event.params["s1c-bind-addr"] log_file = "/tmp/{}.log".format(random.randint(1000, 100000)) command = " ".join( [ "/home/ubuntu/srsLTE/build/srsenb/src/srsenb", "--enb.name=dummyENB01", "--enb.mcc=901", "--enb.mnc=70", "--enb.mme_addr={}".format(mme_addr), "--enb.gtp_bind_addr={}".format(gtp_bind_addr), "--enb.s1c_bind_addr={}".format(s1c_bind_addr), "--enb_files.rr_config=/configzmq/rr.conf", "--enb_files.sib_config=/configzmq/sib.conf", "--enb_files.drb_config=/configzmq/drb.conf", "/configzmq/enb.conf", "--rf.device_name=zmq", "--rf.device_args='fail_on_disconnect=true,tx_port=tcp://*:2000,rx_port=tcp://localhost:2001,id=enb,base_srate=23.04e6'", ] ) logger.debug("Register action: executing") process = self._run_daemon(command, log_file) logger.debug("Register action: executed") event.set_results( {"status": "ok", "pid": process.pid, "log-file": log_file} ) except subprocess.CalledProcessError as e: event.fail("Command error: {}".format(e.output)) except asyncio.TimeoutError as e: event.fail("Timeout error") except Exception as e: event.fail(e) def on_unregister_action(self, event): """Unregister action""" try: command = "sudo killall -s KILL srsenb" output = subprocess.check_output(command, shell=True) event.set_results( { "status": "ok", "message": "Unregistered successfully", "output": output, } ) except subprocess.CalledProcessError as e: event.fail("Command error: {}".format(e.output)) except Exception as e: event.fail(e) def on_attach_ue_action(self, event): """Attach User Emulator to EnodeB.""" try: usim_imsi = event.params["usim-imsi"] usim_k = event.params["usim-k"] usim_opc = event.params["usim-opc"] log_file = "/tmp/{}.log".format(random.randint(1000, 100000)) command = " ".join( [ "sudo", "/home/ubuntu/srsLTE/build/srsue/src/srsue", "--usim.imsi={}".format(usim_imsi), "--usim.k={}".format(usim_k), "--usim.algo=milenage", "--usim.opc={}".format(usim_opc), "--nas.apn=oai.ipv4", "--rf.device_name=zmq", "--rf.device_args='tx_port=tcp://*:2001,rx_port=tcp://localhost:2000,id=ue,base_srate=23.04e6'", "/configzmq/ue.conf", ] ) logger.debug("Attach UE action: executing") process = self._run_daemon(command, log_file) logger.debug("Attach UE action: executed") event.set_results( {"status": "ok", "pid": process.pid, "log-file": log_file} ) except subprocess.CalledProcessError as ex: event.fail(ex) def on_detach_ue_action(self, event): """Detach UE action""" try: command = "sudo killall -s KILL srsue" output = subprocess.check_output(command, shell=True) event.set_results( {"status": "ok", "message": "Detached successfully", "output": output} ) except subprocess.CalledProcessError as e: event.fail("Command error: {}".format(e.output)) except Exception as e: event.fail(e) def on_remove_default_gw_action(self, event): """Remove default gw""" try: command = "sudo route del default" output = subprocess.check_output(command, shell=True) event.set_results( {"status": "ok", "message": "Default route removed!", "output": output} ) except subprocess.CalledProcessError as e: event.fail("Command error: {}".format(e.output)) except Exception as e: event.fail(e) def _run_daemon(self, cmd, stdout_file): with open(stdout_file, "wb") as f: return subprocess.Popen(cmd, shell=True, stdout=f) if __name__ == "__main__": main(EnodebCharm)