AntiVirus を支える情報検索アルゴリズム 2 - Boyer–Moore(BM) 法 & Wu-Manber (WM) 法

static inline cl_error_t matcher_run(
    const struct cli_matcher *root,
    const unsigned char *buffer, uint32_t length,
    const char **virname, struct cli_ac_data *mdata,
    uint32_t offset,
    const struct cli_target_info *tinfo,
    cli_file_t ftype,
    struct cli_matched_type **ftoffset,
    unsigned int acmode,
    unsigned int pcremode,
    struct cli_ac_result **acres,
    fmap_t *map,
    struct cli_bm_off *offdata,
    struct cli_pcre_off *poffdata,
    cli_ctx *ctx
)

class BMSearcher:

    def __init__(self, pattern: str):
        self.pattern = pattern
        self.m = len(pattern)

        # Bad Character Table の作成
        self._bad_char_table: Dict[str, int] = self._build_bad_char_table()

            
    def _build_bad_char_table(self) -> Dict[str, int]:
        # Skip table: Dict[str, int]
        # 検索対象のキーワード内の文字と、その文字がキーワード内で最後に出現する位置のインデックス
        table = {}
        for i, char in enumerate(self.pattern):
            table[char] = i
        return table

@dataclass(frozen=True)
class SearchStep:
    shift: int                      # シフト位置
    mismatch_index: Optional[int]   # 不一致が発生したインデックス(末尾からの相対位置)
    bad_char: Optional[str]         # 不一致となった文字
    shift_amount: int               # 次のステップのためのシフト量
    description: str                # 説明


class BMSearcher:

    def __init__(self, pattern: str):
        self.pattern = pattern
        self.m = len(pattern)

        # Bad Character Table の作成
        self._bad_char_table: Dict[str, int] = self._build_bad_char_table()


    def _build_bad_char_table(self) -> Dict[str, int]:
        # Skip table: Dict[str, int]
        # 検索対象のキーワード内の文字と、その文字がキーワード内で最後に出現する位置のインデックス
        table = {}
        for i, char in enumerate(self.pattern):
            table[char] = i
        return table


    # 検索処理の可視化のため、検索中の状態を返す Generator 関数
    def search_generator(self, text: str) -> Iterator[SearchStep]:

        n = len(text)
        shift = 0

        while shift <= n - self.m:
            j = self.m - 1
            
            # 末尾から順にキーワードとテキストの一致を比較する
            while j >= 0 and self.pattern[j] == text[shift + j]:
                j -= 1

            # 文字列が完全に一致した場合
            if j < 0:
                yield SearchStep(
                    shift=shift,
                    mismatch_index=None,
                    bad_char=None,
                    shift_amount=1,
                    description="Complete match found!"
                )
                shift += 1

            # 検索中に不一致が発生した場合
            else:
                text_char = text[shift + j]
                
                # Bad Character Rule の計算式: shift_amount = max(1, mismatch_index - last_occurrence_index)
                last_occurrence = self._bad_char_table.get(text_char, -1)
                
                # スキップのためのシフト量(ループ回避のため最低 1 はシフトさせる)
                shift_amount = max(1, j - last_occurrence)
                
                reason = (
                    f"Bad char '{text_char}' found at pattern index {last_occurrence}."
                    if last_occurrence != -1 and last_occurrence < j
                    else f"Bad char '{text_char}' logic (not in pattern or to right)."
                )

                yield SearchStep(
                    shift=shift,
                    mismatch_index=j,
                    bad_char=text_char,
                    shift_amount=shift_amount,
                    description=reason
                )
                
                shift += shift_amount

class BMHSearcher:

    def __init__(self, pattern: str):
        self.pattern = pattern
        self.m = len(pattern)
        self._shift_table: Dict[str, int] = self._build_shift_table()

    def _build_shift_table(self) -> Dict[str, int]:

        table = {}
        # パターンの最後の文字以外 (0 から m-2 まで) を登録
        for i in range(self.m - 1):
            char = self.pattern[i]
            table[char] = self.m - 1 - i
        return table

class WuManber:

    def __init__(self, patterns: List[str], block_size: int = 2) -> None:
        if not patterns:
            raise ValueError("Invalid patterns.")
        
        self.patterns = patterns
        self.block_size = block_size
        self.min_len = min(len(p) for p in patterns)

        if self.min_len < self.block_size:
            raise ValueError(f"Block size {self.block_size} is larger than the minimum pattern length {self.min_len}.")

        # SHIFT テーブル
        self.shift_table: Dict[str, int] = {}

        # HASH テーブル
        self.hash_table: Dict[str, List[int]] = defaultdict(list)

        # PREFIX テーブル / Python の組込み関数 hash() は整数を返すのでそのまま利用
        self.prefix_table: List[int] = []

        self._build_tables()

    def _get_hash(self, text_block: str) -> int:
        return hash(text_block)

    def _build_tables(self) -> None:

        for idx, pattern in enumerate(self.patterns):

            prefix_part = pattern[:self.block_size]
            self.prefix_table.append(self._get_hash(prefix_part))

            limit_pattern = pattern[:self.min_len]
            
            # shift_table: {'ap': 3, 'pp': 2, 'pl': 1, 'le': 0, 'am': 3}
            for i in range(self.min_len - self.block_size + 1):
                block = limit_pattern[i : i + self.block_size]
                shift = self.min_len - 1 - (i + self.block_size - 1)
                
                if block not in self.shift_table:
                    self.shift_table[block] = shift
                else:
                    self.shift_table[block] = min(self.shift_table[block], shift)
                
                if shift == 0:
                    self.hash_table[block].append(idx)

def __init__(self, patterns: List[str], block_size: int = 2) -> None:
    if not patterns:
        raise ValueError("Invalid patterns.")

    self.patterns = patterns
    self.block_size = block_size
    self.min_len = min(len(p) for p in patterns)

    if self.min_len < self.block_size:
        raise ValueError(f"Block size {self.block_size} is larger than the minimum pattern length {self.min_len}.")

    # SHIFT テーブル
    self.shift_table: Dict[str, int] = {}

    # HASH テーブル
    self.hash_table: Dict[str, List[int]] = defaultdict(list)

    # PREFIX テーブル / Python の組込み関数 hash() は整数を返すのでそのまま利用
    self.prefix_table: List[int] = []

    self._build_tables()

def _build_tables(self) -> None:

    for idx, pattern in enumerate(self.patterns):

        prefix_part = pattern[:self.block_size]
        self.prefix_table.append(self._get_hash(prefix_part))

        limit_pattern = pattern[:self.min_len]

        for i in range(self.min_len - self.block_size + 1):
            block = limit_pattern[i : i + self.block_size]
            shift = self.min_len - 1 - (i + self.block_size - 1)

            if block not in self.shift_table:
                self.shift_table[block] = shift
            else:
                self.shift_table[block] = min(self.shift_table[block], shift)

            if shift == 0:
                self.hash_table[block].append(idx)

block = limit_pattern[i : i + self.block_size]
shift = self.min_len - 1 - (i + self.block_size - 1)

if block not in self.shift_table:
    self.shift_table[block] = shift
else:
    self.shift_table[block] = min(self.shift_table[block], shift)

if shift == 0:
    self.hash_table[block].append(idx)

def search(self, text: str) -> Iterator[Tuple[int, str]]:
    n = len(text)
    m = self.min_len
    B = self.block_size

    if n < m:
        return

    idx = m - 1
    default_shift = m - B + 1

    while idx < n:
        # 末尾ブロックを取得してシフト量を決定
        suffix_block = text[idx - B + 1 : idx + 1]
        shift = self.shift_table.get(suffix_block, default_shift)

        # シフト量が0の場合、候補パターンと比較(それ以外はシフト)
        if shift == 0:

            candidate_indices = self.hash_table.get(suffix_block, [])
            text_start_pos = idx - m + 1
            text_prefix_block = text[text_start_pos : text_start_pos + B]
            text_prefix_hash = self._get_hash(text_prefix_block)

            for p_idx in candidate_indices:
                if self.prefix_table[p_idx] != text_prefix_hash:
                    continue

                pattern = self.patterns[p_idx]
                p_len = len(pattern)

                if text_start_pos + p_len <= n:
                    if text[text_start_pos : text_start_pos + p_len] == pattern:
                        yield (text_start_pos, pattern)

            idx += 1

        else:
            idx += shift

cl_error_t cli_bm_scanbuff(
    const unsigned char *buffer, 
    uint32_t length, 
    const char **virname, 
    const struct cli_bm_patt **patt, 
    const struct cli_matcher *root, 
    uint32_t offset, 
    const struct cli_target_info *info, 
    struct cli_bm_off *offdata, 
    cli_ctx *ctx
);

struct cli_matcher {
    unsigned int type;

    /* Extended Boyer-Moore */
    uint8_t *bm_shift;
    struct cli_bm_patt **bm_suffix, **bm_pattab;
    uint32_t *soff, soff_len; /* for PE section sigs */
    uint32_t bm_offmode, bm_patterns, bm_reloff_num, bm_absoff_num;

    /* HASH */
    struct cli_hash_patt hm;
    struct cli_hash_wild hwild;

    /* Extended Aho-Corasick */
    uint32_t ac_partsigs, ac_nodes, ac_lists, ac_patterns, ac_lsigs;
    struct cli_ac_lsig **ac_lsigtable;
    struct cli_ac_node *ac_root, **ac_nodetable;
    struct cli_ac_list **ac_listtable;
    struct cli_ac_patt **ac_pattable;
    struct cli_ac_patt **ac_reloff;
    uint32_t ac_reloff_num, ac_absoff_num;
    uint8_t ac_mindepth, ac_maxdepth;
    struct filter *filter;

    uint16_t maxpatlen;
    uint8_t ac_only;

    {省略}
};

#define BM_MIN_LENGTH 3
#define BM_BLOCK_SIZE 3
#define HASH(a, b, c) (211 * a + 37 * b + c)

{省略}

i = BM_MIN_LENGTH - BM_BLOCK_SIZE;
for (; i < length - BM_BLOCK_SIZE + 1;) {
    idx   = HASH(buffer[i], buffer[i + 1], buffer[i + 2]);
    shift = root->bm_shift[idx];

    if (shift == 0) {
        prefix = buffer[i - BM_MIN_LENGTH + BM_BLOCK_SIZE];
        p      = root->bm_suffix[idx];
        if (p && p->cnt == 1 && p->pattern0 != prefix) {
        {省略}