我们从Python开源项目中，提取了以下35个代码示例，用于说明如何使用mmap.PROT_READ。
def remoteSceneChanged(self, data): w, h, size, newfile = data #self._sizeHint = (whint, hhint) if self.shm is None or self.shm.size != size: if self.shm is not None: self.shm.close() if sys.platform.startswith('win'): self.shmtag = newfile ## on windows, we create a new tag for every resize self.shm = mmap.mmap(-1, size, self.shmtag) ## can't use tmpfile on windows because the file can only be opened once. elif sys.platform == 'darwin': self.shmFile.close() self.shmFile = open(self._view.shmFileName(), 'r') self.shm = mmap.mmap(self.shmFile.fileno(), size, mmap.MAP_SHARED, mmap.PROT_READ) else: self.shm = mmap.mmap(self.shmFile.fileno(), size, mmap.MAP_SHARED, mmap.PROT_READ) self.shm.seek(0) data = self.shm.read(w*h*4) self._img = QtGui.QImage(data, w, h, QtGui.QImage.Format_ARGB32) self._img.data = data # data must be kept alive or PySide 1.2.1 (and probably earlier) will crash. self.update()
def wipe(self): filter_bitmap_fd = os.open("/dev/shm/kafl_filter0", os.O_RDWR | os.O_SYNC | os.O_CREAT) os.ftruncate(filter_bitmap_fd, self.config.config_values['BITMAP_SHM_SIZE']) filter_bitmap = mmap.mmap(filter_bitmap_fd, self.config.config_values['BITMAP_SHM_SIZE'], mmap.MAP_SHARED, mmap.PROT_WRITE | mmap.PROT_READ) for i in range(self.config.config_values['BITMAP_SHM_SIZE']): filter_bitmap[i] = '\x00' filter_bitmap.close() os.close(filter_bitmap_fd) filter_bitmap_fd = os.open("/dev/shm/kafl_tfilter", os.O_RDWR | os.O_SYNC | os.O_CREAT) os.ftruncate(filter_bitmap_fd, 0x1000000) filter_bitmap = mmap.mmap(filter_bitmap_fd, 0x1000000, mmap.MAP_SHARED, mmap.PROT_WRITE | mmap.PROT_READ) for i in range(0x1000000): filter_bitmap[i] = '\x00' filter_bitmap.close() os.close(filter_bitmap_fd)
def __set_binary(self, filename, binaryfile, max_size): shm_fd = os.open(filename, os.O_RDWR | os.O_SYNC | os.O_CREAT) os.ftruncate(shm_fd, max_size) shm = mmap.mmap(shm_fd, max_size, mmap.MAP_SHARED, mmap.PROT_WRITE | mmap.PROT_READ) shm.seek(0x0) shm.write('\x00' * max_size) shm.seek(0x0) f = open(binaryfile, "rb") bytes = f.read(1024) if bytes: shm.write(bytes) while bytes != "": bytes = f.read(1024) if bytes: shm.write(bytes) f.close() shm.close() os.close(shm_fd)
def init(self): self.control = socket.socket(socket.AF_UNIX) while True: try: self.control.connect(self.control_filename) #self.control.connect(self.control_filename) break except socket_error: pass #time.sleep(0.01) self.kafl_shm_f = os.open(self.bitmap_filename, os.O_RDWR | os.O_SYNC | os.O_CREAT) self.fs_shm_f = os.open(self.payload_filename, os.O_RDWR | os.O_SYNC | os.O_CREAT) #argv_fd = os.open(self.argv_filename, os.O_RDWR | os.O_SYNC | os.O_CREAT) os.ftruncate(self.kafl_shm_f, self.bitmap_size) os.ftruncate(self.fs_shm_f, (128 << 10)) #os.ftruncate(argv_fd, (4 << 10)) self.kafl_shm = mmap.mmap(self.kafl_shm_f, self.bitmap_size, mmap.MAP_SHARED, mmap.PROT_WRITE | mmap.PROT_READ) self.fs_shm = mmap.mmap(self.fs_shm_f, (128 << 10), mmap.MAP_SHARED, mmap.PROT_WRITE | mmap.PROT_READ) return True
def __init__(self): if not NativeGPIO.instance: GPIOPort.__init__(self, 54) self.export = range(54) self.post_value = True self.post_function = True self.gpio_setup = [] self.gpio_reset = [] self.valueFile = [0 for i in range(54)] self.functionFile = [0 for i in range(54)] for i in range(54): # Export the pins here to prevent a delay when accessing the values for the # first time while waiting for the file group to be set self.__checkFilesystemExport__(i) try: with open('/dev/gpiomem', 'rb') as gpiomem: self.gpio_map = mmap.mmap(gpiomem.fileno(), BLOCK_SIZE, prot=mmap.PROT_READ) except OSError as err: error(err) NativeGPIO.instance = self
def get(self,request): n=12 try: size = os.path.getsize(ACTIVITY_LOG) with open(ACTIVITY_LOG, "rb") as f: # for Windows the mmap parameters are different fm = mmap.mmap(f.fileno(), 0, mmap.MAP_SHARED, mmap.PROT_READ) for i in xrange(size - 1, -1, -1): if fm[i] == '\n': n -= 1 if n == -1: break lines = fm[i + 1 if i else 0:].splitlines() return JsonResponse({'status': "success", 'log': lines}) except Exception as err: return JsonResponse({'status' : "error",'messagge' : err}) finally: try: fm.close() except (UnboundLocalError, TypeError): return JsonResponse({'status':"error", 'message': "Activity log file is empty"}) #index
def __get_shm(self, type_id, slave_id): if slave_id in self.tmp_shm[type_id]: shm = self.tmp_shm[type_id][slave_id] else: shm_fd = os.open(self.files[type_id] + str(slave_id), os.O_RDWR | os.O_SYNC) shm = mmap.mmap(shm_fd, self.sizes[type_id]*self.tasks_per_requests, mmap.MAP_SHARED, mmap.PROT_WRITE | mmap.PROT_READ) self.tmp_shm[type_id][slave_id] = shm return shm
def open_global_bitmap(self): self.global_bitmap_fd = os.open(self.config.argument_values['work_dir'] + "/bitmap", os.O_RDWR | os.O_SYNC | os.O_CREAT) os.ftruncate(self.global_bitmap_fd, self.bitmap_size) self.global_bitmap = mmap.mmap(self.global_bitmap_fd, self.bitmap_size, mmap.MAP_SHARED, mmap.PROT_WRITE | mmap.PROT_READ)
def get_map(cls, size): prot = mmap.PROT_EXEC | mmap.PROT_WRITE | mmap.PROT_READ return cls(-1, size, prot=prot)
def count_line(path): """count lines of text Edited date: 160626 Test: 160626 Examples: :: path = 'test.text' self.count_line(text) Args: path (str): text file Returns: int: number of lines """ f = os.open(path, os.O_RDONLY) buf = mmap.mmap(f, 0, prot=mmap.PROT_READ) answer = 0 readline = buf.readline while readline(): answer += 1 buf.close() return int(answer)
def resize(self, width, height): # Drop previous buffer and shm data if necessary if self.buffer: self.buffer.destroy() self.shm_data.close() wl_shm_format, cairo_shm_format = self._w.shm_formats[0] stride = cairo.ImageSurface.format_stride_for_width( cairo_shm_format, width) size = stride * height with AnonymousFile(size) as fd: self.shm_data = mmap.mmap( fd, size, prot=mmap.PROT_READ | mmap.PROT_WRITE, flags=mmap.MAP_SHARED) pool = self._w.shm.create_pool(fd, size) self.buffer = pool.create_buffer( 0, width, height, stride, wl_shm_format) pool.destroy() self.s = cairo.ImageSurface(cairo_shm_format, width, height, data=self.shm_data, stride=stride) self.surface.attach(self.buffer, 0, 0) self.width = width self.height = height if self.redraw_func: self.redraw_func(self) self.surface.commit()
def keyboard_keymap(self, keyboard, format_, fd, size): print("keyboard_keymap {} {} {}".format(format_, fd, size)) keymap_data = mmap.mmap( fd, 0, prot=mmap.PROT_READ, flags=mmap.MAP_SHARED) os.close(fd) # The provided keymap appears to have a terminating NULL which # xkbcommon chokes on. Specify length=size-1 to remove it. keymap = self._c.xkb_context.keymap_new_from_buffer( keymap_data, length=size - 1) keymap_data.close() self.keyboard_state = keymap.state_new()
def __init__(self, knowledgebase): if _sanity_check: ExprGenerator.setup() State.ExtraInfoGen = ExprGenerator Perceptron.model = Model() Perceptron.indepKB = IndepKnowledgeBase() Perceptron.KB = knowledgebase knowledgebase.postedit_indepkb(Perceptron.indepKB) Perceptron.c = 0 Perceptron.iter = FLAGS.iter Perceptron.beamsize = FLAGS.beam Perceptron.parser = Parser(Perceptron.indepKB, Perceptron.KB, Perceptron.model, State) Perceptron.ncpus = FLAGS.ncpus Perceptron.ontheflyfd = FLAGS.ontheflyfd Perceptron.single_gold = FLAGS.singlegold Perceptron.output_prefix = FLAGS.outputprefix Perceptron.fdbeamsize = FLAGS.fdbeam if Perceptron.ncpus > 0: Perceptron.shared_memo_size = int(1024 * 1024 * 1024) # 1G shared memory Perceptron.shared_memo = mmap.mmap(-1, Perceptron.shared_memo_size, mmap.MAP_SHARED, mmap.PROT_READ | mmap.PROT_WRITE) Perceptron.ref_beams = {} if FLAGS.ref: print >> LOGS, "loading refs", hgs = pickle.load(open(FLAGS.ref)) self.load_hgs(hgs) if FLAGS.extraref: print >> LOGS, "loading extra refs", hgs = pickle.load(open(FLAGS.extraref)) self.load_hgs(hgs)
def __open_qqwry(self): """Open qqwry.dat. """ with open(self.path, "rb") as f: self.file_obj = mmap.mmap( f.fileno(), 0, mmap.MAP_SHARED, mmap.PROT_READ)
def __init__(self, filename): """Read the given TrueType file. :Parameters: `filename` The name of any Windows, OS2 or Macintosh Truetype file. The object must be closed (see `close`) after use. An exception will be raised if the file does not exist or cannot be read. """ if not filename: filename = '' len = os.stat(filename).st_size self._fileno = os.open(filename, os.O_RDONLY) if hasattr(mmap, 'MAP_SHARED'): self._data = mmap.mmap(self._fileno, len, mmap.MAP_SHARED, mmap.PROT_READ) else: self._data = mmap.mmap(self._fileno, len, None, mmap.ACCESS_READ) offsets = _read_offset_table(self._data, 0) self._tables = {} for table in _read_table_directory_entry.array(self._data, offsets.size, offsets.num_tables): self._tables[table.tag] = table self._names = None self._horizontal_metrics = None self._character_advances = None self._character_kernings = None self._glyph_kernings = None self._character_map = None self._glyph_map = None self._font_selection_flags = None self.header = \ _read_head_table(self._data, self._tables['head'].offset) self.horizontal_header = \ _read_horizontal_header(self._data, self._tables['hhea'].offset)
def __init__(self,fname): f = file(fname) header = f.readline() row = f.readline() self.ofs = len(header) self.lline = len(row) self.ldata = len(row.strip()) self.skip = self.lline-self.ldata self.skip_char = row[self.ldata:] self.mmap = mmap.mmap(f.fileno(),0,prot=mmap.PROT_READ)
def __init__(self,fname,split_chrom="",**kwargs): self.logger = logging.getLogger("indexed_fasta") self.fname = fname self.chrom_stats = {} self.split_chrom = split_chrom # try to load index ipath = fname + '.byo_index' if os.access(ipath,os.R_OK): self.load_index(ipath) else: self.index() self.store_index(ipath) f = file(fname) self.mmap = mmap.mmap(f.fileno(),0,prot=mmap.PROT_READ)
def __init__(self,fname,split_chrom="",**kwargs): self.fname = fname self.chrom_stats = {} self.split_chrom = split_chrom # try to load index ipath = fname + '.byo_index' if os.access(ipath,os.R_OK): self.load_index(ipath) else: self.index() self.store_index(ipath) f = file(fname) self.mmap = mmap.mmap(f.fileno(),0,prot=mmap.PROT_READ)
def create_function_from_shellcode(shell_code, restype=ctypes.c_int64, argtypes=()): mm = mmap.mmap(-1, len(shell_code), flags=mmap.MAP_SHARED | mmap.MAP_ANONYMOUS, prot=mmap.PROT_WRITE | mmap.PROT_READ | mmap.PROT_EXEC) mm.write(shell_code) ctypes_buffer = ctypes.c_int.from_buffer(mm) function = ctypes.CFUNCTYPE(restype, *argtypes)(ctypes.addressof(ctypes_buffer)) function._avoid_gc_for_mmap = mm return function
def from_ShellCode(cls, shell_code, restype=ctypes.c_int64, *argtypes): if not isinstance(shell_code, bytes): shell_code = bytes(shell_code, 'utf-8') mm = mmap.mmap(-1, len(shell_code), flags=mmap.MAP_SHARED | mmap.MAP_ANONYMOUS, prot=mmap.PROT_WRITE | mmap.PROT_READ | mmap.PROT_EXEC) mm.write(shell_code) return cls(mm, restype, *argtypes)
def __init__(self, phys_address, size=PAGE_SIZE): """ Create object which maps physical memory to Python's mmap object. :param phys_address: based address of physical memory """ self._size = size phys_address -= phys_address % PAGE_SIZE fd = self._open_dev("/dev/mem") self._memmap = mmap.mmap(fd, size, flags=mmap.MAP_SHARED, prot=mmap.PROT_READ | mmap.PROT_WRITE, offset=phys_address) self._close_dev(fd) atexit.register(self.cleanup)
def get_nuimo_battery_level(mac_address): # pragma: no cover, try: fd = os.open('/tmp/' + mac_address.replace(':', '-'), os.O_RDONLY) except FileNotFoundError: logger.error("No memory mapped file named %s, return None battery level", mac_address.replace(':', '-')) return None buf = mmap.mmap(fd, mmap.PAGESIZE, mmap.MAP_SHARED, mmap.PROT_READ) result, = struct.unpack('i', buf[:4]) return result
def __init__(self, filename): self.filename = abspath(expanduser(filename)) self.file = open(self.filename, "r") self.faiFilename = faiFilename(self.filename) if getsize(self.filename) > 0: self.view = mmap.mmap(self.file.fileno(), 0, prot=mmap.PROT_READ) self.fai = loadFastaIndex(self.faiFilename, self.view) else: self.view = None self.fai = [] self.contigLookup = self._loadContigLookup()
def _mmap(fd, offset, size): prot = MMAP.PROT_READ | MMAP.PROT_WRITE flags = MMAP.MAP_PRIVATE if size &0xfff !=0: size = (size & ~0xfff) + 0x1000 assert size > 0 result = mmap_function(0, size, prot, flags, fd, offset) return ctypes.cast(result, ctypes.POINTER(ctypes.c_char))
def get_urls(self): """Extract all URLs embedded in this file through a simple regex.""" if not os.path.getsize(self.file_path): return [] # http://stackoverflow.com/a/454589 urls = set() f = open(self.file_path, "rb") m = mmap.mmap(f.fileno(), 0, access=mmap.PROT_READ) for url in re.findall(URL_REGEX, m): url = list(url) if not url[2] : m = re.match(IP_REGEX,url[1]) if m is None : tld = "".join(url[1].split(".")[-1:]) while not is_whitelisted_tld (tld) and tld : url[1]=url[1][:-1] tld=tld[:-1] if not tld : continue if not is_whitelisted_domain(url[1]): if not is_whitelisted_url("".join(url)): urls.add("".join(url)) return list(urls)
def fn(self, syscall): if syscall.result is not None: return proc = syscall.process addr = InjectedMemory(syscall.process, 1024, prot=mmap.PROT_READ | mmap.PROT_WRITE | mmap.PROT_EXEC) proc.write_bytes(addr.addr, self.EXIT_CODE) p = syscall.process.tracer.backend.debugger[syscall.process.pid] p.setInstrPointer(addr.addr) regs = p.getregs() regs.orig_rax = SYSCALL_NOOP p.setregs(regs)
def __init__( self, process, length, prot=mmap.PROT_READ | mmap.PROT_WRITE, flags=mmap.MAP_ANONYMOUS | mmap.MAP_PRIVATE): self.length = length self.process = process self.mapped = True self.caller_frame = inspect.stack()[1] parameters = [0, length, prot, flags, -1, 0] self.addr = inject_syscall(process, SYSCALL_MMAP, parameters)
def __get_chunk(self, start=0, end=None): ''' Gets chunk from the sar combo file, from start to end :param start: where to start a pulled chunk :type start: int. :param end: where to end a pulled chunk :type end: int. :return: str. ''' piece = False if (self.__filename and os.access(self.__filename, os.R_OK)): fhandle = None try: fhandle = os.open(self.__filename, os.O_RDONLY) except OSError: print(("Couldn't open file %s" % (self.__filename))) fhandle = None if (fhandle): try: sarmap = mmap.mmap(fhandle, length=0, prot=mmap.PROT_READ) except (TypeError, IndexError): os.close(fhandle) traceback.print_exc() #sys.exit(-1) return False if (not end): end = sarmap.size() try: sarmap.seek(start) piece = sarmap.read(end - start) except: traceback.print_exc() os.close(fhandle) return(piece)
def __split_file(self): ''' Splits combined SAR output file (in ASCII format) in order to extract info we need for it, in the format we want. :return: ``List``-style of SAR file sections separated by the type of info they contain (SAR file sections) without parsing what is exactly what at this point ''' # Filename passed checks through __init__ if (self.__filename and os.access(self.__filename, os.R_OK)): fhandle = None try: fhandle = os.open(self.__filename, os.O_RDONLY) except OSError: print(("Couldn't open file %s" % (self.__filename))) fhandle = None if (fhandle): try: sarmap = mmap.mmap(fhandle, length=0, prot=mmap.PROT_READ) except (TypeError, IndexError): os.close(fhandle) traceback.print_exc() #sys.exit(-1) return False sfpos = sarmap.find(PATTERN_MULTISPLIT, 0) while (sfpos > -1): '''Split by day found''' self.__splitpointers.append(sfpos) # Iterate for new position try: sfpos = sarmap.find(PATTERN_MULTISPLIT, (sfpos + 1)) except ValueError: print("ValueError on mmap.find()") return True if (self.__splitpointers): # Not sure if this will work - if empty set # goes back as True here return True return False
def createEncFile(self, fname, obfuscate, twice, outerKey=None, outerIv=None, innerIv=None): """ read plaintext file, then open, write and save encrypted file @param fname: name of the plaintext file to encrypt @param twice: True if data should be encrypted twice @param outerKey: usually None, if the same file is encrypted twice it is different be default, by design, because the outerKey and outerIv and innerIv are random. If one wants to produce an identical encrypted file multiple time (e.g. for a safetyCheckDec()) then one needs to fix the outerKey and outerIv. @param outerIv: see outerKey @param innerIv: see outerKey """ self.settings.mlogger.log("Opening file %s in order to encrypt it." % (fname), logging.DEBUG, "Encryption") with open(fname, "rb") as f: # Size 0 will read the ENTIRE file into memory! # File is open read-only # mmap does not implement __exit__ so we cannot use "with mmap... as m:" try: m = mmap.mmap(f.fileno(), 0, prot=mmap.PROT_READ) except ValueError as e: self.settings.mlogger.log("Is file '%s' empty? You cannot encrypt " "an empty file. (%s)" % (fname, e), logging.DEBUG, "Encryption") raise ValueError("Is file '%s' empty? You cannot encrypt " "an empty file. (%s)" % (fname, e)) s = m.size() self.blob = m.read(s) del m if len(self.blob) != s: raise IOError("File IO problem - not enough data read") self.settings.mlogger.log("Read %d bytes from file %s." % (s, fname), logging.DEBUG, "Encryption") # encrypt rng = Random.new() self.outerKey = rng.read(KEYSIZE) if outerKey is not None: self.outerKey = outerKey self.versionSw = basics.VERSION_STR self.noOfEncryptions = 1 outputfname = self.saveBlobToEncFile(fname, obfuscate, twice, outerKey, outerIv, innerIv) # overwrite with nonsense to shred memory self.outerKey = rng.read(KEYSIZE) return outputfname # output file name
def fn(self, syscall): if syscall.result is not None: return p = syscall.process.tracer.backend.debugger[syscall.process.pid] proc = syscall.process text = b"hello world!\n" # backup state before syscall orig_ip = p.getInstrPointer() backup = Backup() backup.backup(p.getregs()) # prepare buffer for text that will be written to stdout buffer = InjectedMemory(syscall.process, 1024) proc.write_bytes(buffer.addr, text) # create instructions for write(0, buffer, size) on fly code = b''.join([ self.CODE_SET_SYSCALL, self.CODE_SET_DESCRIPTOR, self.CODE_SET_LEN + struct.pack("<q", len(text)), self.CODE_SET_BUF + struct.pack("<q", buffer.addr), self.CODE_SYSCALL ]) # prepare region for code instructions addr = InjectedMemory(syscall.process, 1024, prot=mmap.PROT_READ | mmap.PROT_WRITE | mmap.PROT_EXEC) proc.write_bytes(addr.addr, code) # jump to injected code p.setInstrPointer(addr.addr) # replace syscall with NOOP syscall regs = p.getregs() regs.orig_rax = SYSCALL_NOOP p.setregs(regs) # execute all instructions that we have written to injected memory # XXX: performance: maybe we can insert syscall(SIGTRAP) while p.getInstrPointer() < addr.addr + len(code): p.singleStep() p.waitEvent() # go back before original syscall p.setInstrPointer(orig_ip - 2) p.syscall() p.waitSyscall() # unmap memory in before state addr.munmap() buffer.munmap() # we are before syscall, restore registers regs = p.getregs() backup.restore(regs) p.setregs(regs)
