org_fedora_oscap.cpioarchive.CpioArchive._readfile() - Code Metrics - Inspection of "Merge the cpio module into the addon." - OpenSCAP/oscap-anaconda-addon - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#140)

by Matěj

created 2021-03-11 16:13 UTC

CpioArchive._readfile() A

↳ Parent: org_fedora_oscap.cpioarchive

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	2
rs	10
c	0
b	0
f	0
cc	1
nop	2

from __future__ import absolute_import

""" cpioarchive: Support for cpio archives
Copyright (C) 2006 Ignacio Vazquez-Abrams """

""" This library is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version.

This library 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 Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA """

import atexit

def version():
  """Returns the version number of the module."""
  return '0.1'

class CpioError(Exception):
  """Exception class for cpioarchive exceptions"""
  pass

class CpioEntry(object):
  """Information about a single file in a cpio archive. Provides a file-like
interface for interacting with the entry."""
  def __init__(self, hdr, cpio, offset):
    """Create a new CpioEntry instance. Internal use only."""
    if len(hdr)<110:
      raise CpioError('cpio header too short')
    if not hdr.startswith(b'070701'):
      raise CpioError('cpio header invalid')
    self.inode=int(hdr[6:14], 16)
    self.mode=int(hdr[14:22], 16)
    self.uid=int(hdr[22:30], 16)
    self.gid=int(hdr[30:38], 16)
    self.nlinks=int(hdr[38:46], 16)
    self.mtime=int(hdr[46:54], 16)
    self.size=int(hdr[54:62], 16)
    """Size of the file stored in the entry."""
    self.devmajor=int(hdr[62:70], 16)
    self.devminor=int(hdr[70:78], 16)
    self.rdevmajor=int(hdr[78:86], 16)
    self.rdevminor=int(hdr[86:94], 16)
    namesize=int(hdr[94:102], 16)
    self.checksum=int(hdr[102:110], 16)
    if len(hdr)<110+namesize:
      raise CpioError('cpio header too short')
    self.name=hdr[110:110+namesize-1].decode("utf-8")
    """Name of the file stored in the entry."""
    self.datastart=offset+110+namesize
    self.datastart+=(4-(self.datastart%4))%4
    self.curoffset=0
    self.cpio=cpio
    self.closed=False

  def close(self):
    """Close this cpio entry. Further calls to methods will raise an
exception."""
    self.closed=True

  def flush(self):
    """Flush the entry (no-op)."""
    pass

  def read(self, size=None):
    """Read data from the entry.

Keyword arguments:
size -- Number of bytes to read (default: whole entry)
"""
    if self.closed:
      raise ValueError('read operation on closed file')
    self.cpio.file.seek(self.datastart+self.curoffset, 0)
    if size and size<self.size-self.curoffset:
      ret=self.cpio.file.read(size)
    else:
      ret=self.cpio.file.read(self.size-self.curoffset)
    self.curoffset+=len(ret)
    return ret

  def seek(self, offset, whence=0):
    """Move to new position within an entry.

Keyword arguments:
offset -- Byte count
whence -- Describes how offset is used.
  0: From beginning of file
  1: Forwards from current position
  2: Backwards from current position
  Other values are ignored.
"""
    if self.closed:
      raise ValueError('seek operation on closed file')
    if whence==0:
      self.curoffset=offset
    elif whence==1:
      self.curoffset+=offset
    elif whence==2:
      self.curoffset-=offset
    self.curoffset=min(max(0, self.curoffset), self.size)

  def tell(self):
    """Get current position within an entry"""
    if self.closed:
      raise ValueError('tell operation on closed file')
    return self.curoffset

class CpioArchive(object):
  @classmethod
  def open(name=None, mode='r', fileobj=None):
    """Open a cpio archive. Defers to CpioArchive.__init__()."""
    return CpioArchive(name, mode, fileobj)

  def __init__(self, name=None, mode='r', fileobj=None):
    """Open a cpio archive.

Keyword arguments:
name -- Filename to open (default: open a file object instead)
mode -- Filemode to open the archive in (default: read-only)
fileobj -- File object to use (default: open by filename instead)
"""
    if not mode=='r':
      raise NotImplementedError()
    self._infos=[]
    if name:
      self._readfile(name)
      self.external=False
    elif fileobj:
      self._readobj(fileobj)
      self.external=True
    else:
      raise CpioError('Oh come on! Pass me something to work with...')
    self._ptr=0
    self.closed=False
    atexit.register(self.close)

  def close(self):
    """Close the CpioArchive. Also closes all associated entries."""
    if self.closed:
      return
    [x.close() for x in self._infos]
    self.closed=True
    if not self.external:
      self.file.close()

  def __next__(self):
    return self.next()

  def next(self):
    """Return the next entry in the archive."""
    if self.closed:
      raise ValueError('next operation on closed file')
    if self._ptr>len(self._infos):
      raise StopIteration()
    ret=self._infos[self._ptr]
    self._ptr+=1
    return ret

  def __iter__(self):
    return iter(self._infos)

  def _readfile(self, name):
    self._readobj(open(name, 'rb'))

  def _readobj(self, fileobj):
    self.file=fileobj
    start=self.file.tell()
    istart=self.file.tell()
    text=self.file.read(110)
    while text:
      namelen=int(text[94:102], 16)
      text+=self.file.read(namelen)
      ce=CpioEntry(text, self, istart)
      if not ce.name=="TRAILER!!!":
        self._infos.append(ce)
      else:
        return
      self.file.seek((4-(self.file.tell()-istart)%4)%4, 1)
      self.file.seek(self._infos[-1].size, 1)
      self.file.seek((4-(self.file.tell()-istart)%4)%4, 1)

      istart=self.file.tell()
      text=self.file.read(110)
    else:
      raise CpioError('premature end of headers')


1			from __future__ import absolute_import
2
3			""" cpioarchive: Support for cpio archives
4			Copyright (C) 2006 Ignacio Vazquez-Abrams """
5
6			""" This library is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 2.1 of the License, or (at your option) any later version.
7
8			This library 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 Lesser General Public License for more details.
9
10			You should have received a copy of the GNU Lesser General Public License along with this library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA """
11
12			import atexit
13
14			def version():
15			"""Returns the version number of the module."""
16			return '0.1'
17
18			class CpioError(Exception):
19			"""Exception class for cpioarchive exceptions"""
20			pass
21
22			class CpioEntry(object):
23			"""Information about a single file in a cpio archive. Provides a file-like
24			interface for interacting with the entry."""
25			def __init__(self, hdr, cpio, offset):
26			"""Create a new CpioEntry instance. Internal use only."""
27			if len(hdr)<110:
28			raise CpioError('cpio header too short')
29			if not hdr.startswith(b'070701'):
30			raise CpioError('cpio header invalid')
31			self.inode=int(hdr[6:14], 16)
32			self.mode=int(hdr[14:22], 16)
33			self.uid=int(hdr[22:30], 16)
34			self.gid=int(hdr[30:38], 16)
35			self.nlinks=int(hdr[38:46], 16)
36			self.mtime=int(hdr[46:54], 16)
37			self.size=int(hdr[54:62], 16)
38			"""Size of the file stored in the entry."""
39			self.devmajor=int(hdr[62:70], 16)
40			self.devminor=int(hdr[70:78], 16)
41			self.rdevmajor=int(hdr[78:86], 16)
42			self.rdevminor=int(hdr[86:94], 16)
43			namesize=int(hdr[94:102], 16)
44			self.checksum=int(hdr[102:110], 16)
45			if len(hdr)<110+namesize:
46			raise CpioError('cpio header too short')
47			self.name=hdr[110:110+namesize-1].decode("utf-8")
48			"""Name of the file stored in the entry."""
49			self.datastart=offset+110+namesize
50			self.datastart+=(4-(self.datastart%4))%4
51			self.curoffset=0
52			self.cpio=cpio
53			self.closed=False
54
55			def close(self):
56			"""Close this cpio entry. Further calls to methods will raise an
57			exception."""
58			self.closed=True
59
60			def flush(self):
61			"""Flush the entry (no-op)."""
62			pass
63
64			def read(self, size=None):
65			"""Read data from the entry.
66
67			Keyword arguments:
68			size -- Number of bytes to read (default: whole entry)
69			"""
70			if self.closed:
71			raise ValueError('read operation on closed file')
72			self.cpio.file.seek(self.datastart+self.curoffset, 0)
73			if size and size<self.size-self.curoffset:
74			ret=self.cpio.file.read(size)
75			else:
76			ret=self.cpio.file.read(self.size-self.curoffset)
77			self.curoffset+=len(ret)
78			return ret
79
80			def seek(self, offset, whence=0):
81			"""Move to new position within an entry.
82
83			Keyword arguments:
84			offset -- Byte count
85			whence -- Describes how offset is used.
86			0: From beginning of file
87			1: Forwards from current position
88			2: Backwards from current position
89			Other values are ignored.
90			"""
91			if self.closed:
92			raise ValueError('seek operation on closed file')
93			if whence==0:
94			self.curoffset=offset
95			elif whence==1:
96			self.curoffset+=offset
97			elif whence==2:
98			self.curoffset-=offset
99			self.curoffset=min(max(0, self.curoffset), self.size)
100
101			def tell(self):
102			"""Get current position within an entry"""
103			if self.closed:
104			raise ValueError('tell operation on closed file')
105			return self.curoffset
106
107			class CpioArchive(object):
108			@classmethod
109			def open(name=None, mode='r', fileobj=None):
110			"""Open a cpio archive. Defers to CpioArchive.__init__()."""
111			return CpioArchive(name, mode, fileobj)
112
113			def __init__(self, name=None, mode='r', fileobj=None):
114			"""Open a cpio archive.
115
116			Keyword arguments:
117			name -- Filename to open (default: open a file object instead)
118			mode -- Filemode to open the archive in (default: read-only)
119			fileobj -- File object to use (default: open by filename instead)
120			"""
121			if not mode=='r':
122			raise NotImplementedError()
123			self._infos=[]
124			if name:
125			self._readfile(name)
126			self.external=False
127			elif fileobj:
128			self._readobj(fileobj)
129			self.external=True
130			else:
131			raise CpioError('Oh come on! Pass me something to work with...')
132			self._ptr=0
133			self.closed=False
134			atexit.register(self.close)
135
136			def close(self):
137			"""Close the CpioArchive. Also closes all associated entries."""
138			if self.closed:
139			return
140			[x.close() for x in self._infos]
141			self.closed=True
142			if not self.external:
143			self.file.close()
144
145			def __next__(self):
146			return self.next()
147
148			def next(self):
149			"""Return the next entry in the archive."""
150			if self.closed:
151			raise ValueError('next operation on closed file')
152			if self._ptr>len(self._infos):
153			raise StopIteration()
154			ret=self._infos[self._ptr]
155			self._ptr+=1
156			return ret
157
158			def __iter__(self):
159			return iter(self._infos)
160
161			def _readfile(self, name):
162			self._readobj(open(name, 'rb'))
163
164			def _readobj(self, fileobj):
165			self.file=fileobj
166			start=self.file.tell()
167			istart=self.file.tell()
168			text=self.file.read(110)
169			while text:
170			namelen=int(text[94:102], 16)
171			text+=self.file.read(namelen)
172			ce=CpioEntry(text, self, istart)
173			if not ce.name=="TRAILER!!!":
174			self._infos.append(ce)
175			else:
176			return
177			self.file.seek((4-(self.file.tell()-istart)%4)%4, 1)
178			self.file.seek(self._infos[-1].size, 1)
179			self.file.seek((4-(self.file.tell()-istart)%4)%4, 1)
180
181			istart=self.file.tell()
182			text=self.file.read(110)
183			else:
184			raise CpioError('premature end of headers')
185

OpenSCAP / oscap-anaconda-addon

Pull Request — master (#140)

CpioArchive._readfile() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like