"""
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 .
Copyright © 2019 Cloud Linux Software Inc.
This software is also available under ImunifyAV commercial license,
see
"""
import abc
import json
import os
from collections import defaultdict
from dataclasses import dataclass, field
from typing import List, NamedTuple, Optional
from .db import (
DB,
PatchDependencyMatch,
VersionMatch,
HashState,
DefinitionType,
)
from .utils import HashCalculator, get_base_dir
class FileIdentifier(NamedTuple):
rel_path: str
hash: str
vuln_id: Optional[int] = None
vuln_type: Optional[int] = None
# already-fixed (post-patch) fingerprint; recognizes a satisfied member.
checksum: Optional[str] = None
@dataclass
class VersionIdentifier:
id: int
hash: str
file_identifiers: list[FileIdentifier]
# one identifier can match multiple base_dirs, need to keep track of them to avoid duplicate scanning
matched_base_dirs: set = field(default_factory=set)
@dataclass
class PatchDependency:
files: list[FileIdentifier]
# base_dir -> {rel_path: is_vulnerable} for recognized members (matched
# `hash` or `checksum`). Allocated lazily on first match and reset to None
# at finalize, so the (vast majority of) never-matched definitions in a
# large sig set hold no dict rather than an empty one.
matched_by_base_dir: Optional[dict] = None
@dataclass
class HashDefinition:
type: DefinitionType
id: int
hash: str
state: HashState
class Matcher(abc.ABC):
def __init__(self, input_file: str):
self.dict_of_identifiers = self._parse_input(input_file)
@abc.abstractmethod
def _parse_input(self, file_path: str) -> dict[str, list[tuple]]:
pass
class VersionsMatcher(Matcher):
@staticmethod
def _parse_path_hash_pairs(file_hashes: str) -> list[FileIdentifier]:
# accepts file_hashes string like [||]*n
# returns list of FileIdentifier objects
parts = file_hashes.strip().split("|")
return [
FileIdentifier(rel_path, hash_)
for rel_path, hash_ in zip(parts[1::2], parts[::2])
]
def _parse_line(self, line: str) -> Optional[VersionIdentifier]:
# each line is made up as :::[file_hashes]
# and is a list of "||" pairs
state, id_, hash_, file_hashes = line.strip().split(":")
if state != "+":
return None
return VersionIdentifier(
id_, hash_, self._parse_path_hash_pairs(file_hashes)
)
def _parse_input(
self, file_path: str
) -> dict[str, list[VersionIdentifier]]:
# reads file version_identifiers with contents like
# +:10831:38ed3878c51c61af938cd4fd9228b23b:ad8d2ec0797fbe584a2f5c1e0985b188|classes/Product.php|e890fa7432bbe7bee4dcbbff1009ca4b|app/AppKernel.php|
plugins_identifiers_by_path: dict[
str, list[VersionIdentifier]
] = defaultdict(list)
with open(file_path, "r") as file:
for line in file:
if new_identifier := self._parse_line(line):
plugins_identifiers_by_path[
new_identifier.file_identifiers[0].rel_path
].append(new_identifier)
return plugins_identifiers_by_path
def has_full_match(
self,
plugin_identifier: VersionIdentifier,
base_dir: str,
hash_calculator: HashCalculator,
) -> bool:
# 1) check that all files from file_identifiers exist in their paths relative to base_dir
for file_identifier in plugin_identifier.file_identifiers:
if not os.path.isfile(
os.path.join(base_dir, file_identifier.rel_path)
):
return False
# 2) all files exist, now check their hashes
for file_identifier in plugin_identifier.file_identifiers:
if (
hash_calculator.calc_hash(
os.path.join(base_dir, file_identifier.rel_path),
apply_normalization=True,
)
!= file_identifier.hash
):
return False
return True
def match_and_save(
self,
full_path: str,
relative_path: str,
db: DB,
hash_calculator: HashCalculator,
):
is_matched = False
# check if we have any version_identifier matching given path
for plugin_identifier in self.dict_of_identifiers.get(
relative_path, []
):
base_dir = get_base_dir(full_path, relative_path)
# skip if we already have matched this base_dir with this plugin_identifier
if (
base_dir not in plugin_identifier.matched_base_dirs
and self.has_full_match(
plugin_identifier, base_dir, hash_calculator
)
):
plugin_identifier.matched_base_dirs.add(base_dir)
db.versions_matches.buffered_insert(
VersionMatch(
id=plugin_identifier.id,
path=base_dir,
hash=plugin_identifier.hash,
)
)
is_matched = True
return is_matched
class PatchDependenciesMatcher(Matcher):
def _parse_input(self, file_path: str) -> dict[str, list[PatchDependency]]:
# read patch_dependencies file
# each line represent a patch dependency and is made of a list of FileToPatch objects, like:
# +[{"hash": "(...)", "checksum": "(...)", "vulnerability_type": 10, "vulnerability_id": 4346, \
# "filename": "popup-builder/com/helpers/AdminHelper.php"}, \
# {"hash": "(...)", "checksum": "(...)", "vulnerability_type": 10, "vulnerability_id": 4347, \
# "filename": "popup-builder/com/classes/Ajax.php"}]
# we should consider only those lines starting with "+"
patch_deps: dict[str, list[PatchDependency]] = defaultdict(list)
# Each PatchDependency is registered under every member's rel_path, so
# finalize() must iterate this unique list, not dict_of_identifiers.
self.dependencies: list[PatchDependency] = []
with open(file_path, "r") as file:
for line in file:
state, data = line[0], line[1:]
if state != "+":
continue
patch_dependency = PatchDependency(
files=[
FileIdentifier(
rel_path=_["filename"],
hash=_["hash"],
vuln_id=_["vulnerability_id"],
vuln_type=_["vulnerability_type"],
checksum=_.get("checksum"),
)
for _ in json.loads(data)
]
)
self.dependencies.append(patch_dependency)
for file_identifier in patch_dependency.files:
patch_deps[file_identifier.rel_path].append(
patch_dependency
)
return patch_deps
def match_and_save(
self,
full_path: str,
relative_path: str,
file_hash: Optional[str],
) -> bool:
# Record which members of each patch dependency are *recognized* at
# this file. A member is recognized when the file at its path matches
# the member's vulnerable `hash` (a detection) OR its already-fixed
# `checksum` (satisfies the set but is not itself reported). Whether
# the whole set is met (every member recognized) can only be decided
# after the full traversal, so rows are emitted later in finalize().
#
# `file_hash` is the raw content hash of `full_path`, computed once per
# file in main(); it is the hash of the file at `relative_path` within
# this base_dir (os.path.join(base_dir, relative_path) == full_path by
# get_base_dir's definition), so no re-reading/re-hashing is needed.
is_matched = False
for patch_dependency in self.dict_of_identifiers.get(
relative_path, []
):
base_dir = get_base_dir(full_path, relative_path)
for file_identifier in patch_dependency.files:
if file_identifier.rel_path != relative_path:
continue
if file_hash == file_identifier.hash:
is_vulnerable = True
is_matched = True
elif (
file_identifier.checksum
and file_hash == file_identifier.checksum
):
is_vulnerable = False
else:
continue
by_base = patch_dependency.matched_by_base_dir
if by_base is None:
by_base = patch_dependency.matched_by_base_dir = {}
by_base.setdefault(base_dir, {})[relative_path] = is_vulnerable
return is_matched
def finalize(self, db: DB) -> None:
"""Emit patch-dependency detections after the full traversal.
A dependency's members are spread across different files, so
dependencies_met (every member recognized via `hash` or `checksum` in
the same base_dir) can only be decided once the whole scan has seen
them. Only vulnerable (hash-matched) members are reported; already-fixed
(checksum-matched) members satisfy the set and yield no rows.
"""
for patch_dependency in self.dependencies:
recognized_by_dir = patch_dependency.matched_by_base_dir
if not recognized_by_dir:
# never matched during the walk (no dict was allocated)
continue
by_rel = {fi.rel_path: fi for fi in patch_dependency.files}
member_count = len(patch_dependency.files)
for base_dir, recognized in recognized_by_dir.items():
dependencies_met = len(recognized) == member_count
for rel_path, is_vulnerable in recognized.items():
if not is_vulnerable:
continue
file_identifier = by_rel[rel_path]
db.patch_dependencies.buffered_insert(
PatchDependencyMatch(
os.path.join(base_dir, rel_path),
file_identifier.hash,
file_identifier.vuln_id,
file_identifier.vuln_type,
dependencies_met=dependencies_met,
)
)
patch_dependency.matched_by_base_dir = None
class HashesMatcher:
def __init__(self, hashes_file: str):
# Index definitions by digest for O(1) lookup per file, mirroring the
# version/patch matchers' dict_of_identifiers. _parse_input keeps the
# malware/superseded filtering, so only active records are indexed.
self.records_by_hash: dict[str, list[HashDefinition]] = {}
for record in self._parse_input(hashes_file):
self.records_by_hash.setdefault(record.hash, []).append(record)
@staticmethod
def _parse_input(hashes_path: str) -> List[HashDefinition]:
"""
Parses the hashes file and returns a list of HashDefinition,
filtering out malware-related types and state==2.
The lines look like :::
Example: 2:675:ab43f2f7ad32404e1b923a8387f1a167:2
Where :code:`type` can be one of the following:
* DEFINITION_TYPE_MALWARE = 1
* DEFINITION_TYPE_VULNERABILITY = 2
* DEFINITION_TYPE_APPLICATION = 3
* DEFINITION_TYPE_DRYRUN = 4
* DEFINITION_TYPE_MALWARE_RULE = 7
* DEFINITION_TYPE_MALWARE_RULE_DRYRUN = 8
* DEFINITION_TYPE_VULNERABILITY_ECOMMERCE = 9
* DEFINITION_TYPE_VULNERABILITY_PLUGIN = 10
"""
MALWARE_TYPES = {
DefinitionType.MALWARE.value,
DefinitionType.MALWARE_RULE.value,
DefinitionType.MALWARE_RULE_DRYRUN.value,
}
result = []
with open(hashes_path, "r") as f:
for line in f:
parts = line.strip().split(":")
if len(parts) != 4:
continue
typ, id_, hash_, state = parts
typ_int = int(typ)
state_int = int(state)
if (
typ_int in MALWARE_TYPES
or state_int == HashState.SUPERSEDED.value
):
continue
result.append(
HashDefinition(
type=DefinitionType(typ_int),
id=int(id_),
hash=hash_,
state=HashState(state_int),
)
)
return result
def match_and_save(self, file_path, file_hash, db):
is_matched = False
# A file's digest can match several definitions (same hash, different
# vuln_id/type/state); emit a row for each so no detection is dropped.
# Dedup of exact-duplicate rows is delegated to the DB (UNIQUE +
# INSERT OR IGNORE), so no in-RAM state is kept.
for record in self.records_by_hash.get(file_hash, ()):
db.hashes_matches.buffered_insert(
(
file_path,
file_hash,
record.type.value,
record.id,
record.state.value,
)
)
is_matched = True
return is_matched