Python：ctypes可散列c_char陣列替換，不會在'\ 0'字節上跳過

Question

爲了便於說明，此腳本創建一個文件mapfile，該文件包含作爲參數給出的文件內容，前綴爲帶有sha1校驗和的二進制頭，允許在後續運行中進行重複檢測。

這裏需要的是一個可散列的ctypes.c_char替換，它可以使sha1校驗和保持最小模糊，但不會阻塞'\ 0'字節。

# -*- coding: utf8 -* 

import io 
import mmap 
import ctypes 
import hashlib 
import logging 

from collections import OrderedDict 

log = logging.getLogger(__file__) 

def align(size, alignment): 
    """return size aligned to alignment""" 
    excess = size % alignment 
    if excess: 
     size = size - excess + alignment 
    return size 


class Header(ctypes.Structure): 
    Identifier = b'HEAD' 
    _fields_ = [ 
     ('id', ctypes.c_char * 4), 
     ('hlen', ctypes.c_uint16), 
     ('plen', ctypes.c_uint32), 
     ('name', ctypes.c_char * 128), 
     ('sha1', ctypes.c_char * 20), 
    ] 
HeaderSize = ctypes.sizeof(Header) 

class CtsMap: 
    def __init__(self, ctcls, mm, offset = 0): 
     self.ctcls = ctcls 
     self.mm = mm 
     self.offset = offset 

    def __enter__(self): 
     mm = self.mm 
     offset = self.offset 
     ctsize = ctypes.sizeof(self.ctcls) 
     if offset + ctsize > mm.size(): 
      newsize = align(offset + ctsize, mmap.PAGESIZE) 
      mm.resize(newsize) 
     self.ctinst = self.ctcls.from_buffer(mm, offset) 
     return self.ctinst 

    def __exit__(self, exc_type, exc_value, exc_traceback): 
     del self.ctinst 
     self.ctinst = None 

class MapFile: 
    def __init__(self, filename): 
     try: 
      # try to create initial file 
      mapsize = mmap.PAGESIZE 
      self._fd = open(filename, 'x+b') 
      self._fd.write(b'\0' * mapsize) 
     except FileExistsError: 
      # file exists and is writable 
      self._fd = open(filename, 'r+b') 
      self._fd.seek(0, io.SEEK_END) 
      mapsize = self._fd.tell() 
     # mmap this file completely 
     self._fd.seek(0) 
     self._mm = mmap.mmap(self._fd.fileno(), mapsize) 
     self._offset = 0 
     self._toc = OrderedDict() 
     self.gen_toc() 

    def gen_toc(self): 
     while self._offset < self._mm.size(): 
      with CtsMap(Header, self._mm, self._offset) as hd: 
       if hd.id == Header.Identifier and hd.hlen == HeaderSize: 
        self._toc[hd.sha1] = self._offset 
        log.debug('toc: [%s]%s: %s', len(hd.sha1), hd.sha1, self._offset) 
        self._offset += HeaderSize + hd.plen 
       else: 
        break 
      del hd 

    def add_data(self, datafile, data): 
     datasize = len(data) 
     sha1 = hashlib.sha1() 
     sha1.update(data) 
     digest = sha1.digest() 

     if digest in self._toc: 
      log.debug('add_data: %s added already', digest) 
      return None 

     log.debug('add_data: %s, %s bytes, %s', datafile, datasize, digest) 
     with CtsMap(Header, self._mm, self._offset) as hd: 
      hd.id = Header.Identifier 
      hd.hlen = HeaderSize 
      hd.plen = datasize 
      hd.name = datafile 
      hd.sha1 = digest 
     del hd 
     self._offset += HeaderSize 

     log.debug('add_data: %s', datasize) 
     blktype = ctypes.c_char * datasize 
     with CtsMap(blktype, self._mm, self._offset) as blk: 
      blk.raw = data 
     del blk 
     self._offset += datasize 
     return HeaderSize + datasize 

    def close(self): 
     self._mm.close() 
     self._fd.close() 


if __name__ == '__main__': 
    import os 
    import sys 

    logconfig = dict(
     level = logging.DEBUG, 
     format = '%(levelname)5s: %(message)s', 
    ) 
    logging.basicConfig(**logconfig) 

    mf = MapFile('mapfile') 
    for datafile in sys.argv[1:]: 
     if os.path.isfile(datafile): 
      try: 
       data = open(datafile, 'rb').read() 
      except OSError: 
       continue 
      else: 
       mf.add_data(datafile.encode('utf-8'), data) 
    mf.close() 

運行：python3 hashable_ctypes_bytes.py somefiles*

調用它第二次，它通過讀取該文件收集在一個有序字典的所有項目，使用SHA1摘要關鍵。不幸的是，c_char數組語義有點有線，因爲它也行爲像'\ 0'終止的C字符串，導致在這裏截斷校驗和。

見線3和4：

DEBUG: toc: [20]b'\xcd0\xd7\xd3\xbf\x9f\xe1\xfe\xffr\xa6g#\xee\xf8\x84\xb5S,u': 0 
DEBUG: toc: [20]b'\xe9\xfe\x1a;i\xcdG0\x84\x1b\r\x7f\xf9\x14\x868\xbdVl\x8d': 1273 
DEBUG: toc: [19]b'\xa2\xdb\xff$&\xfe\x0f\xb4\xcaB<F\x92\xc0\xf1`(\x96N': 3642 
DEBUG: toc: [15]b'O\x1b~c\x82\xeb)\x8f\xb5\x9c\x15\xd5e:\xa9': 4650 
DEBUG: toc: [20]b'\x80\xe9\xbcF\x97\xdc\x93DG\x90\x19\x8c\xca\xfep\x05\xbdM\xfby': 13841 
DEBUG: add_data: b'\xcd0\xd7\xd3\xbf\x9f\xe1\xfe\xffr\xa6g#\xee\xf8\x84\xb5S,u' added already 
DEBUG: add_data: b'\xe9\xfe\x1a;i\xcdG0\x84\x1b\r\x7f\xf9\x14\x868\xbdVl\x8d' added already 
DEBUG: add_data: b'../python/tmp/colorselect.py', 848 bytes, b'\xa2\xdb\xff$&\xfe\x0f\xb4\xcaB<F\x92\xc0\xf1`(\x96N\x00' 
DEBUG: add_data: 848 
DEBUG: add_data: b'../python/tmp/DemoCode.py', 9031 bytes, b'O\x1b~c\x82\xeb)\x8f\xb5\x9c\x15\xd5e:\xa9\x00p\x0f\xc04' 
DEBUG: add_data: 9031 
DEBUG: add_data: b'\x80\xe9\xbcF\x97\xdc\x93DG\x90\x19\x8c\xca\xfep\x05\xbdM\xfby' added already 

通常的建議是與c_byte取代c_char * 20 * 20，並失去透明字節處理上的方式。除了數據轉換的麻煩之外，由於是字節數組，因此c_byte數組不可散列。我還沒有找到一個實用的解決方案，而不必經歷重複的轉換問題，或者訴諸於使用十六進制數字，這使得文摘大小要求翻了一番。

我認爲，將c_char設計與C零終止語義混合的決定首先是一個錯誤。爲了解決這個問題，我可以想象爲ctypes添加一個c_char_nz類型，可以解決這個問題。

對於那些仔細閱讀代碼的人，您可能會想到ctypes結構的del語句。討論它可以發現here:。

Answer 1

雖然下面的代碼正在做你提到的來回轉換，但它確實很好地隱藏了問題。我發現一個包含null的散列，現在可以將該字段用作字典鍵。希望能幫助到你。

from ctypes import * 
import hashlib 

class Test(Structure): 
    _fields_ = [('_sha1',c_ubyte * 20)] 

    @property 
    def sha1(self): 
     return bytes(self._sha1) 

    @sha1.setter 
    def sha1(self, value): 
     self._sha1 = (c_ubyte * 20)(*value) 

test = Test() 
test.sha1 = hashlib.sha1(b'aaaaaaaaaaa').digest() 
D = {test.sha1:0} 
print(D) 

輸出：

{b'u\\\x00\x1fJ\xe3\xc8\x84>ZP\xddj\xa2\xfa#\x89=\xd3\xad': 0}