elasticsearch模糊查询图片与文件

2019-12-10

0x1 目标

需求是可以在亿级文件或图片中, 快速找到相似的图片或文件

0x2 解决方案

中间尝试过不少方案, 这里就说一下最后测试可行的方案

图片模糊查询: 使用image-match, 这个工程是自己实现的汉明矩阵数据储存和查询, 之前有使用插件形式去解析汉明hash, 但是数据量太大查询太慢, 所以使用了这个方案, 使用空间换时间

ssdeep

1	文件模糊查询: 文件模糊查询使用的ssdeep模糊hash算法

0x3 在es中使用image-match

以下是测试用例, 详细的用法也说了, 有独特的订制需求可以在elasticsearch_driver.py和signature_database_base.py文件中修改

import os
import hashlib
from elasticsearch import Elasticsearch, ConnectionError, RequestError, NotFoundError
from image_match.elasticsearch_driver import SignatureES


DOC_TYPE = 'image'
MAPPINGS = {
  "mappings": {
    DOC_TYPE: {
      "dynamic": True,
      "properties": {
        "metadata": {
            "type": "object",
            "dynamic": True,
            "properties": {
                "tenant_id": { "type": "string", "index" : "not_analyzed"   }
            }
        }
      }
    }
  }
}

def setup_index():
    es = Elasticsearch()
    try:
        es.indices.create(index="images",  body=MAPPINGS)
    except RequestError as e:
        if e.error == u'index_already_exists_exception':
            es.indices.delete(index_name)
        else:
            raise

#建立索引
setup_index()

#插入图片数据
es = Elasticsearch()
ses = SignatureES(es)

#参数可以是url, 路径, 原始图片数据
ses.add_image('xxx.jpg')

ses.add_image('xxx1.jpg')

#检索
r = ses.search_image('xxx.jpg')
print(r)

#查询结果,  dist越小, 说明相似度越高
'''
[
 {'dist': 0.0,
  'id': u'AVM37oZq0osmmAxpPvx7',
  'metadata': None,
  'path': u'xxx.jpg',

  'score': 7.937254},
 {'dist': 0.22095170140933634,
  'id': u'AVM37nMg0osmmAxpPvx6',
  'metadata': None,
  'path': u'xxx1.jpg',

  'score': 0.28797293}

]

'''

0x4 在es中查询ssdeep

ssdeep哈希的格式如下：


    chunksize:chunk:double_chunk

    96:4mkbL7ruouQiayH6Zaw2wL/ThZy9snQzzvPpsplhgHRofZxOfYVHjnU/r2CydUXe:qnNiOZQvvaWAV7VYr68XWvth

但是比较ssdeep的方式, 只能调用ssdeep库的compare函数去比较相似度, 但用在es中, 只能使用写插件的形式, 之前也说了, 在图标模糊搜索的时候尝试过使用插件来搜索结果, 结果是太慢了, 所以这里采用了几个办法去过滤搜索结果

1
2

1. 过滤hash中chunksize相同的
2. 过滤chunk和double_chunk中字符超过7位以上相同的(因为相似的文件的模糊hash会有连续相同的字符, 这个需要在es创建index的时候修改mapping中的分词器, 后面代码中会列出)

具体实现的核心代码:

## es 5.x 创建分词器的, 动态mapping

dsl = {
    "mappings": {
        "self": {
            "_all": {
                "enabled": True
            },
            "dynamic": "true",
            "properties": {
                "file_ssdeep.chunk": { #chunck

                    "analyzer": "ssdeep_analyzer", #指定分词器

                    "type": "text"
                },
                "file_ssdeep.chunk_size": { #size

                    "type": "integer"
                },
                "file_ssdeep.double_chunk": { #double_chunck

                    "analyzer": "ssdeep_analyzer", #指定分词器

                    "type": "text"
                }
            }
        }
    },
    "settings": {
        "analysis": { #自定义分词器

            "analyzer": {
                "ssdeep_analyzer": { #分词器名称

                    "tokenizer": "ssdeep_tokenizer"
                }
            },
            "tokenizer": {
                "ssdeep_tokenizer": { #分词器参数

                    "max_gram": 7, #最大分词数

                    "min_gram": 7, #最小分词数

                    "type": "ngram"
                }
            }
        }
    }
}

#创建index
es = Elasticsearch(['localhost:9200'])  # body = dsl,
result = es.indices.create(index='ssdeep-index', body=dsl, ignore=400)

#插入ssdeep数据

document = {'chunksize': 96, 'chunk': '4mkbL7ruouQiayH6Zaw2wL/ThZy9snQzzvPpsplhgHRofZxOfYVHjnU/r2CydUXe', 'double_chunk': 'qnNiOZQvvaWAV7VYr68XWvth'}

es.index('ssdeep-index', 'record', document)
es.indices.refresh('ssdeep-index')



## 查询ssdeep
ssdeep_value = "96:4mkbL7ruouQiayH6Zaw2wL/ThZy9snQzzvPpsplhgHRofZxOfYVHjnU/r2CydUXe:qnNiOZQvvaWAV7VYr68XWvth" #查询目标
chunksize, chunk, double_chunk = ssdeep_value.split(':')
chunksize = int(chunksize)
#查询语句
query = {
    'query': {
        'bool': {
            'must': [
                {
                    'terms': { #必须是相同的chunksize

                        'chunksize': [chunksize]

                    }
                },
                {
                    'bool': {
                        'should': [ #查询chunk和doublechunk, 因为在创建index时候指定的分词器了, 所以这里的逻辑就是: 如果chunk和doublechunk中有任意一个7个字符以上相同, 则条件成立

                            {
                                'match': {
                                    'chunk': { 

                                        'query': chunk
                                    }
                                }
                            },
                            {
                                'match': {
                                    'double_chunk': {
                                        'query': double_chunk
                                    }
                                }
                            }
                        ],
                        'minimum_should_match': 1 #以上条件任意一个成立, 逻辑成立

                    }
                }
            ]
        }
    }
}

results = es.search('ssdeep-index', body=query)
#相似度 1 - 100, 越高相似度越高
threshold_grade = 80
#相似文件
sim_file = []

for record in results['hits']['hits']:
    #取出结果中的ssdeep
    record_ssdeep = record['_source']['ssdeep']
    #调用库函数比较相似度
    ssdeep_grade = ssdeep.compare(record_ssdeep, ssdeep_value)
    #判断相似度阈值, 是否符合要求
    if ssdeep_grade >= threshold_grade:
        sim_file.append(record)


print('result : ' +  str(sim_file))

0x5 结语

以上就是粗略的记录了此次解决方案的思路, 可能细节还有遗漏, 或者有更好的解决方案可以留言或加V一起交流^-^