Admin

name	type	size	perms	modified	actions
[parent directory]
__pycache__/	dir	-	0755	2026-08-19 08:39:37
fs_scanner/	dir	-	0755	2026-08-19 08:39:37
rpc/	dir	-	0755	2026-08-19 08:39:37
__init__.py	file	0 B	0644	2026-08-05 00:00:00	edit
config.py	file	1,908 B	0644	2026-08-05 00:00:00	edit
constants.py	file	908 B	0644	2026-08-05 00:00:00	edit
custom_integration.py	file	6,216 B	0644	2026-08-05 00:00:00	edit
exceptions.py	file	826 B	0644	2026-08-05 00:00:00	edit
feature.py	file	3,742 B	0644	2026-08-05 00:00:00	edit
license.py	file	5,417 B	0644	2026-08-05 00:00:00	edit
model.py	file	2,055 B	0644	2026-08-05 00:00:00	edit
state.py	file	921 B	0644	2026-08-05 00:00:00	edit
utils.py	file	1,443 B	0644	2026-08-05 00:00:00	edit

editing: license.py

"""
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 <https://www.gnu.org/licenses/>.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import asyncio
import logging
import ssl
import urllib
from contextlib import suppress
from enum import IntEnum
from pathlib import Path
from typing import List, Optional

from defence360agent.utils import importer
from imav.patchman.config import PatchmanConfig

# use lazy import to reduce memory consumption when not in use
x509 = importer.LazyImport("cryptography.x509")
asn1_decoder = importer.LazyImport("pyasn1.codec.der.decoder")

logger = logging.getLogger(__name__)

LICENSE_ADDRESS = "licensing.patchman.co"
LICENSE_PORT = 443
CA_ROOT_FILE = "/etc/patchman/ca.crt"
LICENSE_CERT_FILE = "/etc/patchman/license/patchman.crt"
LICENSE_KEY_FILE = "/etc/patchman/license/patchman.key"
IMUNIFY_KEY_URL_TEMPLATE = "https://{address}:{port}/v2/imunifyav_key/"
FEATURES_EXTENSION_OID = "1.3.6.1.4.1.44098.2.1.3"

class Feature(IntEnum):
    """
    Patchman functionality depending on the license.
    See patchman agent implementation for more details.
    """

Core = 0
    CoreMalwareQuarantine = 1
    CoreVulnerabilitiesPatch = 2
    RulesMalwareScan = 3
    RulesMalwareClean = 4
    EcommerceVulnerabilitiesScan = 5
    EcommerceVulnerabilitiesPatch = 6
    PluginVulnerabilitiesScan = 7
    PluginVulnerabilitiesPatch = 8
    ScanningMultithreaded = 13
    ScanningRealtime = 14
    PatchDependencies = 16
    Waf = 17

class PatchmanLicenseError(Exception):
    pass

class License:
    @classmethod
    def is_active(cls):
        return Path(LICENSE_CERT_FILE).exists()

@classmethod
    def get_features(cls) -> List[Feature]:
        """
        Return the list of features available for the current patchman license
        """
        features = []
        if cls.is_active():
            with open(LICENSE_CERT_FILE, "rb") as f:
                cert = x509.load_pem_x509_certificate(f.read())
            if ext := next(
                (
                    ext
                    for ext in cert.extensions
                    if ext.oid.dotted_string == FEATURES_EXTENSION_OID
                ),
                None,
            ):
                asn1_data, _ = asn1_decoder.decode(ext.value.value)
                # functional settings are specified as a bitstring
                bitstring = asn1_data.asBinary()
                # FeatureCore always present
                features.append(Feature.Core)
                for pos, value in enumerate(bitstring, start=1):
                    if value == "1":
                        # ignore unknown or deprecated features
                        with suppress(ValueError):
                            features.append(Feature(pos))
        logger.info("Patchman license features: %s", features)
        return features

@classmethod
    def has_clean_mode(cls) -> bool:
        return Feature.RulesMalwareClean in cls.get_features()

@classmethod
    def has_realtime_enabled(cls):
        return Feature.ScanningRealtime in cls.get_features()

@classmethod
    def _get_imunify_key(cls) -> Optional[str]:
        """
        Get imunify registration key from the Patchman license service.
        See LICENSING-84 for details.
        """
        cafile = PatchmanConfig.get("licensing", "ca", default=CA_ROOT_FILE)
        address = PatchmanConfig.get(
            "licensing", "address", default=LICENSE_ADDRESS
        )
        port = PatchmanConfig.get("licensing", "port", default=LICENSE_PORT)
        context = ssl.create_default_context()
        context.load_verify_locations(cafile=cafile)
        context.load_cert_chain(
            certfile=LICENSE_CERT_FILE,
            keyfile=LICENSE_KEY_FILE,
        )
        with urllib.request.urlopen(
            IMUNIFY_KEY_URL_TEMPLATE.format(address=address, port=port),
            context=context,
        ) as response:
            status = response.status
            if response.status == 200:  # key found
                return response.read().decode()
            elif status == 404:  # key was not found
                return None
            raise PatchmanLicenseError(
                f"Patchman license service returns {response.status} response"
            )

@classmethod
    async def get_imunify_key(cls) -> Optional[str]:
        loop = asyncio.get_event_loop()
        try:
            return await asyncio.wait_for(
                loop.run_in_executor(None, cls._get_imunify_key),
                timeout=300,
            )
        except FileNotFoundError as exc:
            logger.warning("Can't get imunify key due to %s", exc)
        except Exception as exc:
            logger.error("Can't get imunify key due to %s", exc)
        return None

cancel