"""
NETCONF for APK Distributions server:
adelie-services module, OpenRC edition
Copyright © 2020 Adélie Software in the Public Benefit, Inc.
Released under the terms of the NCSA license. See the LICENSE file included
with this source distribution for more information.
SPDX-License-Identifier: NCSA
"""
import errno
import logging
import os
import pathlib
import subprocess
from netconf import error, util
from ncserver.base.service import Service, ServiceStatus
from ncserver.base.util import _, yang_dt_for_timestamp
LOGGER = logging.getLogger(__name__)
"""The object used for logging informational messages."""
M_ABI_VERSION = 1
"""The ABI version of this NETCONF module."""
M_PREFIX = "svcs"
"""The XML tag prefix for this module's tags."""
M_NS = "http://netconf.adelielinux.org/ns/service"
"""The XML namespace for this module."""
M_NAME = "adelie-services"
"""The YANG model name for this module."""
M_REVISION = "2020-09-22"
"""The YANG revision date for this module."""
M_IMPORTS = {
'ietf-yang-types@2013-07-15': {
'ns': "urn:ietf:params:xml:ns:yang:ietf-yang-types", 'prefix': "yang"
}
}
"""The imported YANG modules for this module."""
def valid_name(name):
"""Determine if the given name is a valid OpenRC service name."""
valid = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz1234567890_-."
return all(char in valid for char in name)
def is_enabled(name):
"""Determine if this OpenRC service is enabled."""
runlevels = (pathlib.Path('/etc/runlevels/boot'),
pathlib.Path('/etc/runlevels/sysinit'),
pathlib.Path('/etc/runlevels/default'))
for level in runlevels:
for service in level.iterdir():
# Resolve the symlink. If it's the service name, we have a match.
if pathlib.Path(service).resolve().name == name:
return True
return False
def get_var(shell_like, var_name: str, default="") -> str:
"""Retrieve the value of var_name from a shell script-like file shell_like.
:param shell_like:
The path to the shell script-like file.
:param str var_name:
The name of the variable to extract.
:param str default:
The default value to return if the variable is not set.
"""
script = ". /etc/init.d/functions.sh;(. {shlike} && printf %s \"${var}\")"
script = script.format(shlike=shell_like, var=var_name)
proc = subprocess.run(
['/bin/sh', '-e', '-c', script], check=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
)
value = proc.stdout.decode('utf-8')
if value == "":
return default
return value
def check_alive(pidfile) -> bool:
"""Determine if the PID contained in a pidfile is alive."""
pid = 0
try:
with open(pidfile) as fhandle:
pid = int(fhandle.read().rstrip())
except OSError as ose:
if ose.errno == errno.EPERM or ose.errno == errno.EACCES:
return True # the pidfile cannot be accessed
return False # sounds bad to me
except ValueError:
return False # the pidfile doesn't contain a PID
if pid <= 0:
return False
try:
os.kill(pid, 0)
except OSError as ose:
# If EPERM, we can't send the signal because it's alive.
return ose.errno == errno.EPERM
else:
return True
class OpenRCService(Service):
"""The OpenRC implementation of the Service class."""
def __init__(self, name):
"""Initialise the structure."""
super().__init__()
if not valid_name(name):
raise NameError(_("{}: invalid service name").format(name))
svcpath = "/etc/init.d/{name}".format(name=name)
if not os.path.exists(svcpath):
raise ValueError(_("{} does not exist").format(name))
self._name = name
self._description = get_var(svcpath, "description", name)
self._enabled = is_enabled(name)
# Update status and start-time.
self.status()
def status(self):
"""Retrieve the service's status."""
stat = ServiceStatus.Stopped
path = None
status_dirs = {
ServiceStatus.Starting: "/run/openrc/starting",
ServiceStatus.Running: "/run/openrc/started",
ServiceStatus.Stopping: "/run/openrc/stopping"
}
for value, directory in status_dirs.items():
for service in pathlib.Path(directory).iterdir():
if pathlib.Path(service).resolve().name == self.name:
stat = value
path = service
break
if stat == ServiceStatus.Running:
# Check if crashed
for service in pathlib.Path("/run/openrc/daemons").iterdir():
svcdir = pathlib.Path(service)
if svcdir.name == self.name:
for daemon in svcdir.iterdir():
dpid = get_var(daemon, 'pidfile')
if dpid and not check_alive(dpid):
return ServiceStatus.Crashed
# Update start-time
self._start_time = pathlib.Path(path).lstat().st_mtime
return stat
def start(self):
"""Start the service."""
if self.status() in (ServiceStatus.Starting, ServiceStatus.Running):
return
subprocess.Popen(["/sbin/rc-service", self.name, "start"])
def stop(self):
"""Stop the service."""
subprocess.Popen(["/sbin/rc-service", self.name, "stop"])
def reload(self):
"""Reload the service's configuration, if possible."""
def restart(self, full: bool):
"""Restart the service.
OpenRC does not support a concept of full restart, so this parameter
is always ignored.
"""
subprocess.Popen(["/sbin/rc-service", self.name, "restart"])
def __str__(self):
return self.name
def __repr__(self):
return "<OpenRC service {name} ({status})>".format(
name=self.name, status=self.status()
)
def service_list():
"""Return the list of services available on this device."""
services = list()
for service in pathlib.Path('/etc/init.d').iterdir():
if service.suffix == '.sh':
continue
services.append(OpenRCService(service.name))
return services
def info(service: str) -> OpenRCService:
"""Return the Service object associated with the specified service.
:param str service:
The name of the service to look up.
:returns:
The Service object, or None if a service with that name does not exist.
"""
try:
return OpenRCService(service)
except ValueError:
return None
def running(node):
"""Retrieve the service configuration for this device."""
svcs = util.subelm(node, 'svcs:services')
for service in service_list():
svcnode = util.subelm(svcs, 'svcs:service')
svcnode.append(util.leaf_elm('svcs:name', service.name))
svcnode.append(util.leaf_elm('svcs:enabled', service.enabled))
def operational(node):
"""Retrieve the service state for this device."""
svcs = util.subelm(node, 'svcs:services')
for service in service_list():
svcnode = util.subelm(svcs, 'svcs:service')
svcnode.append(util.leaf_elm('svcs:name', service.name))
svcnode.append(util.leaf_elm('svcs:description', service.description))
svcnode.append(util.leaf_elm('svcs:enabled', service.enabled))
svcnode.append(util.leaf_elm(
'svcs:status', service.status().name.lower()
))
if service.status() == ServiceStatus.Running:
svcnode.append(util.leaf_elm(
'svcs:start-time', yang_dt_for_timestamp(service.start_time)
))
def edit(session, rpc, node, def_op):
"""Edit the service configuration for this device."""
