Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
ObjectFormatException in case of parsing error (malformed
field date)
Returns: Tuple of (author, time, (timezone, timezone_neg_utc))
"""
try:
sep = value.rindex(b'> ')
except ValueError:
return (value, None, (None, False))
try:
person = value[0:sep+1]
rest = value[sep+2:]
timetext, timezonetext = rest.rsplit(b' ', 1)
time = int(timetext)
timezone, timezone_neg_utc = parse_timezone(timezonetext)
except ValueError as e:
raise ObjectFormatException(e)
return person, time, (timezone, timezone_neg_utc)
:param text: Serialized text to parse
:return: iterator of tuples of (name, mode, sha)
:raise ObjectFormatException: if the object was malformed in some way
"""
count = 0
l = len(text)
while count < l:
mode_end = text.index(b' ', count)
mode_text = text[count:mode_end]
if strict and mode_text.startswith(b'0'):
raise ObjectFormatException("Invalid mode '%s'" % mode_text)
try:
mode = int(mode_text, 8)
except ValueError:
raise ObjectFormatException("Invalid mode '%s'" % mode_text)
name_end = text.index(b'\0', mode_end)
name = text[mode_end+1:name_end]
count = name_end+21
sha = text[name_end+1:count]
if len(sha) != 20:
raise ObjectFormatException("Sha has invalid length")
hexsha = sha_to_hex(sha)
yield (name, mode, hexsha)
self._check_has_member("_commit_time", "missing commit time")
for parent in self._parents:
check_hexsha(parent, "invalid parent sha")
check_hexsha(self._tree, "invalid tree sha")
check_identity(self._author, "invalid author")
check_identity(self._committer, "invalid committer")
check_time(self._author_time)
check_time(self._commit_time)
last = None
for field, _ in _parse_message(self._chunked_text):
if field == _TREE_HEADER and last is not None:
raise ObjectFormatException("unexpected tree")
elif field == _PARENT_HEADER and last not in (_PARENT_HEADER,
_TREE_HEADER):
raise ObjectFormatException("unexpected parent")
elif field == _AUTHOR_HEADER and last not in (_TREE_HEADER,
_PARENT_HEADER):
raise ObjectFormatException("unexpected author")
elif field == _COMMITTER_HEADER and last != _AUTHOR_HEADER:
raise ObjectFormatException("unexpected committer")
elif field == _ENCODING_HEADER and last != _COMMITTER_HEADER:
raise ObjectFormatException("unexpected encoding")
last = field
def _apply_pack(self, refs):
all_exceptions = (IOError, OSError, ChecksumMismatch, ApplyDeltaError,
AssertionError, socket.error, zlib.error,
ObjectFormatException)
status = []
will_send_pack = False
for command in refs:
if command[1] != ZERO_SHA:
will_send_pack = True
if will_send_pack:
# TODO: more informative error messages than just the exception
# string
try:
recv = getattr(self.proto, "recv", None)
self.repo.object_store.add_thin_pack(self.proto.read, recv)
status.append((b'unpack', b'ok'))
except all_exceptions as e:
status.append((b'unpack', str(e).replace('\n', '')))
check_identity(self._committer, "invalid committer")
last = None
for field, _ in _parse_message(self._chunked_text):
if field == _TREE_HEADER and last is not None:
raise ObjectFormatException("unexpected tree")
elif field == _PARENT_HEADER and last not in (_PARENT_HEADER,
_TREE_HEADER):
raise ObjectFormatException("unexpected parent")
elif field == _AUTHOR_HEADER and last not in (_TREE_HEADER,
_PARENT_HEADER):
raise ObjectFormatException("unexpected author")
elif field == _COMMITTER_HEADER and last != _AUTHOR_HEADER:
raise ObjectFormatException("unexpected committer")
elif field == _ENCODING_HEADER and last != _COMMITTER_HEADER:
raise ObjectFormatException("unexpected encoding")
last = field
def check_hexsha(hex, error_msg):
"""Check if a string is a valid hex sha string.
:param hex: Hex string to check
:param error_msg: Error message to use in exception
:raise ObjectFormatException: Raised when the string is not valid
"""
if not valid_hexsha(hex):
raise ObjectFormatException("%s %s" % (error_msg, hex))
def _parse_legacy_object(self, map):
"""Parse a legacy object, setting the raw string."""
text = _decompress(map)
header_end = text.find(b'\0')
if header_end < 0:
raise ObjectFormatException("Invalid object header, no \\0")
self.set_raw_string(text[header_end+1:])
self._check_has_member("_object_sha", "missing object sha")
self._check_has_member("_object_class", "missing object type")
self._check_has_member("_name", "missing tag name")
if not self._name:
raise ObjectFormatException("empty tag name")
check_hexsha(self._object_sha, "invalid object sha")
if getattr(self, "_tagger", None):
check_identity(self._tagger, "invalid tagger")
last = None
for field, _ in _parse_message(self._chunked_text):
if field == _OBJECT_HEADER and last is not None:
raise ObjectFormatException("unexpected object")
elif field == _TYPE_HEADER and last != _OBJECT_HEADER:
raise ObjectFormatException("unexpected type")
elif field == _TAG_HEADER and last != _TYPE_HEADER:
raise ObjectFormatException("unexpected tag name")
elif field == _TAGGER_HEADER and last != _TAG_HEADER:
raise ObjectFormatException("unexpected tagger")
last = field
def check_identity(identity, error_msg):
"""Check if the specified identity is valid.
This will raise an exception if the identity is not valid.
Args:
identity: Identity string
error_msg: Error message to use in exception
"""
email_start = identity.find(b'<')
email_end = identity.find(b'>')
if (email_start < 0 or email_end < 0 or email_end <= email_start
or identity.find(b'<', email_start + 1) >= 0
or identity.find(b'>', email_end + 1) >= 0
or not identity.endswith(b'>')):
raise ObjectFormatException(error_msg)
def _deserialize(self, chunks):
"""Grab the metadata attached to the tag"""
self._tagger = None
self._tag_time = None
self._tag_timezone = None
self._tag_timezone_neg_utc = False
for field, value in _parse_message(chunks):
if field == _OBJECT_HEADER:
self._object_sha = value
elif field == _TYPE_HEADER:
obj_class = object_class(value)
if not obj_class:
raise ObjectFormatException("Not a known type: %s" % value)
self._object_class = obj_class
elif field == _TAG_HEADER:
self._name = value
elif field == _TAGGER_HEADER:
(self._tagger,
self._tag_time,
(self._tag_timezone,
self._tag_timezone_neg_utc)) = parse_time_entry(value)
elif field is None:
if value is None:
self._message = None
self._signature = None
else:
try:
sig_idx = value.index(BEGIN_PGP_SIGNATURE)
except ValueError: