calculate-utils-3-lib/pym/calculate/lib/utils/system.py

# -*- coding: utf-8 -*-

# Copyright 2017 Mir Calculate. http://www.calculate-linux.org
#
#  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
from .files import grepFile, getRunCommands
from .tools import cached, traverse
from os import path
import os
from collections import defaultdict
from fnmatch import fnmatch, filter as fnfilter

_ = lambda x: x
from ..cl_lang import setLocalTranslate

setLocalTranslate('cl_lib3', sys.modules[__name__])

class X86():
    def __init__(self, rootdn):
        self.rootdn = rootdn
        self.usrlib = path.join(rootdn, "usr/lib")
        self.lib = path.join(rootdn, "lib")
        self.cusrlib = path.join(rootdn, "usr/lib")
        self.clib = path.join(rootdn, "lib")
        self.libmodules = path.join(rootdn, "lib/modules")
        self.pythonsite = path.join(rootdn, "usr/lib/python3.10/site-packages")
        self.libopengl = path.join(rootdn, "usr/lib/opengl")

class Amd64():
    def __init__(self, rootdn):
        self.rootdn = rootdn
        self.usrlib = path.join(rootdn, "usr/lib64")
        self.lib = path.join(rootdn, "lib64")
        self.cusrlib = path.join(rootdn, "usr/lib")
        self.clib = path.join(rootdn, "lib")
        self.libmodules = path.join(rootdn, "lib/modules")
        self.pythonsite = path.join(rootdn, "usr/lib64/python3.10/site-packages")
        self.libopengl = path.join(rootdn, "usr/lib64/opengl")

@cached(each_instance=True)
def SystemPath(rootdn):
    lib64path = path.join(rootdn, "lib64")
    libpath = path.join(rootdn, "lib")
    if path.exists(lib64path):
        return Amd64(rootdn)
    else:
        return X86(rootdn)


class SystemType():
    """
    Тип контейнера текущей системы
    """
    Uml = "uml"
    VServer = "vserver"
    OpenVZ = "openvz"
    LXC = "lxc"
    Docker = "docker"
    Xen0 = "xen0"
    XenU = "xenU"
    Rkt = "rkt"
    SystemdNSpawn = "systemd-nspawn"
    NotDetected = ""

    @classmethod
    def detect_container(cls):
        if grepFile("/proc/cpuinfo", "UML"):
            return cls.Uml
        elif grepFile("/proc/self/status",
                "(s_context|VxID):\s*[1-9]"):
            return cls.VServer
        elif (path.exists("/proc/vz/veinfo")
              and not path.exists("/proc/vz/version")):
            return cls.OpenVZ
        elif grepFile("/proc/self/status", "envID:\s*[1-9]"):
            return cls.OpenVZ #old test
        elif grepFile("/proc/1/environ", "container=lxc"):
            return cls.LXC
        elif grepFile("/proc/1/environ", "container=rkt"):
            return cls.Rkt
        elif grepFile("/proc/1/environ", "container=systemd-nspawn"):
            return cls.SystemdNSpawn
        elif path.exists("/.dockerenv"):
            return cls.Docker
        elif grepFile("/proc/1/environ", "container=docker"):
            return cls.Docker
        return cls.NotDetected

    @classmethod
    def detect_vm(cls):
        if path.exists("/proc/xen"):
            if grepFile("/proc/xen/capabilities", "control_d"):
                return cls.Xen0
            return cls.XenU
        return cls.NotDetected

    @classmethod
    def detect(cls):
        return cls.detect_container() or cls.detect_vm()

class FUser():
    def __init__(self, ffilter=None):
        self.ffilter = ffilter
        self.ref = defaultdict(list)
        self.refresh()

    def get_mnt(self, pid):
        try:
            return os.readlink("/proc/%s/ns/mnt" %pid).strip()
        except Exception:
            return ""

    def get_main_mnt(self):
        return self.get_mnt("1")

    def refresh(self):
        self.ref=defaultdict(list)
        main_mnt = self.get_main_mnt()
        for p in os.listdir("/proc/"):
            if not p.isdigit(): continue
            if self.get_mnt(p) != main_mnt: continue
            d = "/proc/%s/fd/" % p
            try:
               for fd in os.listdir(d):
                 f = os.readlink(d+fd)
                 self.ref[f].append(p)
            except OSError:
                pass

    def search(self, item):
        keys = fnfilter(self.ref.keys(), item)
        return traverse(self.ref[k] for k in keys)

    def __contains__(self, item):
        return item in self.ref

    def __getitem__(self, item):
        return self.ref[item]

emerge_running = lambda: any(fnmatch(x, "*python-exec/*/emerge*")
                             for x in getRunCommands(True))