This is an archive of the discontinued Mercurial Phabricator instance.

Differential D7881

lfs: check content length after downloading content
ClosedPublic

Authored by mharbison72 on Jan 14 2020, 11:09 PM.

Details

Reviewers
pulkit
Group Reviewers
hg-reviewers
Commits
rHG0ee0a3f6a990: lfs: check content length after downloading content
Summary

Adapted from the Facebook repo[1]. The intent is to distinguish between the
connection dying and getting served a corrupt blob.

The original message:

HTTP makes no provision to tell your client that you failed halfway through
producing your response and won't have the answer they're looking for. So, if a
LFS server fails while producing a response, then we'll report an OID mismatch.

We can do a little better and disambiguate between "the server sent us the
wrong blob" (very scary) and "the server crashed" (merely annoying) by looking
at the content length of the response we got back. If it's not what was
advertised, we can reasonably safely assume the server crashed.

[1] https://github.com/facebookexperimental/eden/commit/2a4a6fab4e882ed89b948bfc1e7d56d7c3c99dd2

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Revision Contents
Changeset List

Diff 19291

hgext/lfs/blobstore.py

# populate the local store. For an init + push of a local clone, # populate the local store. For an init + push of a local clone,
# the usercache is the only place it _could_ be. If not present, the # the usercache is the only place it _could_ be. If not present, the
# missing file msg here will indicate the local repo, not the usercache. # missing file msg here will indicate the local repo, not the usercache.
if self.cachevfs.exists(oid): if self.cachevfs.exists(oid):
return self.cachevfs(oid, b'rb') return self.cachevfs(oid, b'rb')
return self.vfs(oid, b'rb') return self.vfs(oid, b'rb')
def download(self, oid, src): def download(self, oid, src, content_length):
"""Read the blob from the remote source in chunks, verify the content, """Read the blob from the remote source in chunks, verify the content,
and write to this local blobstore.""" and write to this local blobstore."""
sha256 = hashlib.sha256() sha256 = hashlib.sha256()
size = 0
with self.vfs(oid, b'wb', atomictemp=True) as fp: with self.vfs(oid, b'wb', atomictemp=True) as fp:
for chunk in util.filechunkiter(src, size=1048576): for chunk in util.filechunkiter(src, size=1048576):
fp.write(chunk) fp.write(chunk)
sha256.update(chunk) sha256.update(chunk)
size += len(chunk)
# If the server advertised a length longer than what we actually
# received, then we should expect that the server crashed while
# producing the response (but the server has no way of telling us
# that), and we really don't need to try to write the response to
# the localstore, because it's not going to match the expected.
if content_length is not None and int(content_length) != size:
msg = (
b"Response length (%s) does not match Content-Length "
b"header (%d): likely server-side crash"
)
raise LfsRemoteError(_(msg) % (size, int(content_length)))
realoid = node.hex(sha256.digest()) realoid = node.hex(sha256.digest())
if realoid != oid: if realoid != oid:
raise LfsCorruptionError( raise LfsCorruptionError(
_(b'corrupt remote lfs object: %s') % oid _(b'corrupt remote lfs object: %s') % oid
) )
self._linktousercache(oid) self._linktousercache(oid)
request.add_header('Content-Length', len(request.data)) request.add_header('Content-Length', len(request.data))
for k, v in headers: for k, v in headers:
request.add_header(pycompat.strurl(k), pycompat.strurl(v)) request.add_header(pycompat.strurl(k), pycompat.strurl(v))
response = b'' response = b''
try: try:
with contextlib.closing(self.urlopener.open(request)) as res: with contextlib.closing(self.urlopener.open(request)) as res:
contentlength = res.info().get(b"content-length")
ui = self.ui # Shorten debug lines ui = self.ui # Shorten debug lines
if self.ui.debugflag: if self.ui.debugflag:
ui.debug(b'Status: %d\n' % res.status) ui.debug(b'Status: %d\n' % res.status)
# lfs-test-server and hg serve return headers in different # lfs-test-server and hg serve return headers in different
# order # order
headers = pycompat.bytestr(res.info()).strip() headers = pycompat.bytestr(res.info()).strip()
ui.debug(b'%s\n' % b'\n'.join(sorted(headers.splitlines()))) ui.debug(b'%s\n' % b'\n'.join(sorted(headers.splitlines())))
if action == b'download': if action == b'download':
# If downloading blobs, store downloaded data to local # If downloading blobs, store downloaded data to local
# blobstore # blobstore
localstore.download(oid, res) localstore.download(oid, res, contentlength)
else: else:
while True: while True:
data = res.read(1048576) data = res.read(1048576)
if not data: if not data:
break break
response += data response += data
if response: if response:
ui.debug(b'lfs %s response: %s' % (action, response)) ui.debug(b'lfs %s response: %s' % (action, response))
for p in _deduplicate(pointers): for p in _deduplicate(pointers):
content = fromstore.read(p.oid(), verify=True) content = fromstore.read(p.oid(), verify=True)
with self.vfs(p.oid(), b'wb', atomictemp=True) as fp: with self.vfs(p.oid(), b'wb', atomictemp=True) as fp:
fp.write(content) fp.write(content)
def readbatch(self, pointers, tostore): def readbatch(self, pointers, tostore):
for p in _deduplicate(pointers): for p in _deduplicate(pointers):
with self.vfs(p.oid(), b'rb') as fp: with self.vfs(p.oid(), b'rb') as fp:
tostore.download(p.oid(), fp) tostore.download(p.oid(), fp, None)
class _nullremote(object): class _nullremote(object):
"""Null store storing blobs to /dev/null.""" """Null store storing blobs to /dev/null."""
def __init__(self, repo, url): def __init__(self, repo, url):
pass pass

hgext/lfs/wireprotolfsserver.py

existed = localstore.has(oid) existed = localstore.has(oid)
# TODO: how to handle timeouts? The body proxy handles limiting to # TODO: how to handle timeouts? The body proxy handles limiting to
# Content-Length, but what happens if a client sends less than it # Content-Length, but what happens if a client sends less than it
# says it will? # says it will?
statusmessage = hgwebcommon.statusmessage statusmessage = hgwebcommon.statusmessage
try: try:
localstore.download(oid, req.bodyfh) localstore.download(oid, req.bodyfh, req.headers[b'Content-Length'])
res.status = statusmessage(HTTP_OK if existed else HTTP_CREATED) res.status = statusmessage(HTTP_OK if existed else HTTP_CREATED)
except blobstore.LfsCorruptionError: except blobstore.LfsCorruptionError:
_logexception(req) _logexception(req)
# XXX: Is this the right code? # XXX: Is this the right code?
res.status = statusmessage(422, b'corrupt blob') res.status = statusmessage(422, b'corrupt blob')
# There's no payload here, but this is the header that lfs-test-server # There's no payload here, but this is the header that lfs-test-server

tests/test-lfs-serve-access.t

> >
> def reposetup(ui, repo): > def reposetup(ui, repo):
> # Nothing to do with a remote repo > # Nothing to do with a remote repo
> if not repo.local(): > if not repo.local():
> return > return
> >
> store = repo.svfs.lfslocalblobstore > store = repo.svfs.lfslocalblobstore
> class badstore(store.__class__): > class badstore(store.__class__):
> def download(self, oid, src): > def download(self, oid, src, contentlength):
> '''Called in the server to handle reading from the client in a > '''Called in the server to handle reading from the client in a
> PUT request.''' > PUT request.'''
> origread = src.read > origread = src.read
> def _badread(nbytes): > def _badread(nbytes):
> # Simulate bad data/checksum failure from the client > # Simulate bad data/checksum failure from the client
> return b'0' * len(origread(nbytes)) > return b'0' * len(origread(nbytes))
> src.read = _badread > src.read = _badread
> super(badstore, self).download(oid, src) > super(badstore, self).download(oid, src, contentlength)
> >
> def _read(self, vfs, oid, verify): > def _read(self, vfs, oid, verify):
> '''Called in the server to read data for a GET request, and then > '''Called in the server to read data for a GET request, and then
> calls self._verify() on it before returning.''' > calls self._verify() on it before returning.'''
> global _readerr > global _readerr
> # One time simulation of a read error > # One time simulation of a read error
> if _readerr: > if _readerr:
> _readerr = False > _readerr = False
$LOCALIP - - [$ERRDATE$] HG error: Exception happened while processing request '/.git/info/lfs/objects/batch': (glob) $LOCALIP - - [$ERRDATE$] HG error: Exception happened while processing request '/.git/info/lfs/objects/batch': (glob)
$LOCALIP - - [$ERRDATE$] HG error: Traceback (most recent call last): (glob) $LOCALIP - - [$ERRDATE$] HG error: Traceback (most recent call last): (glob)
$LOCALIP - - [$ERRDATE$] HG error: verifies = store.verify(oid) (glob) $LOCALIP - - [$ERRDATE$] HG error: verifies = store.verify(oid) (glob)
$LOCALIP - - [$ERRDATE$] HG error: raise IOError(errno.EIO, r'%s: I/O error' % oid.decode("utf-8")) (glob) $LOCALIP - - [$ERRDATE$] HG error: raise IOError(errno.EIO, r'%s: I/O error' % oid.decode("utf-8")) (glob)
$LOCALIP - - [$ERRDATE$] HG error: *Error: [Errno *] b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c: I/O error (glob) $LOCALIP - - [$ERRDATE$] HG error: *Error: [Errno *] b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c: I/O error (glob)
$LOCALIP - - [$ERRDATE$] HG error: (glob) $LOCALIP - - [$ERRDATE$] HG error: (glob)
$LOCALIP - - [$ERRDATE$] HG error: Exception happened while processing request '/.hg/lfs/objects/b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c': (glob) $LOCALIP - - [$ERRDATE$] HG error: Exception happened while processing request '/.hg/lfs/objects/b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c': (glob)
$LOCALIP - - [$ERRDATE$] HG error: Traceback (most recent call last): (glob) $LOCALIP - - [$ERRDATE$] HG error: Traceback (most recent call last): (glob)
$LOCALIP - - [$ERRDATE$] HG error: localstore.download(oid, req.bodyfh) (glob) $LOCALIP - - [$ERRDATE$] HG error: localstore.download(oid, req.bodyfh, req.headers[b'Content-Length'])
$LOCALIP - - [$ERRDATE$] HG error: super(badstore, self).download(oid, src) (glob) $LOCALIP - - [$ERRDATE$] HG error: super(badstore, self).download(oid, src, contentlength)
$LOCALIP - - [$ERRDATE$] HG error: _(b'corrupt remote lfs object: %s') % oid (glob) $LOCALIP - - [$ERRDATE$] HG error: _(b'corrupt remote lfs object: %s') % oid (glob)
$LOCALIP - - [$ERRDATE$] HG error: LfsCorruptionError: corrupt remote lfs object: b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c (no-py3 !) $LOCALIP - - [$ERRDATE$] HG error: LfsCorruptionError: corrupt remote lfs object: b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c (no-py3 !)
$LOCALIP - - [$ERRDATE$] HG error: hgext.lfs.blobstore.LfsCorruptionError: corrupt remote lfs object: b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c (py3 !) $LOCALIP - - [$ERRDATE$] HG error: hgext.lfs.blobstore.LfsCorruptionError: corrupt remote lfs object: b5bb9d8014a0f9b1d61e21e796d78dccdf1352f23cd32812f4850b878ae4944c (py3 !)
$LOCALIP - - [$ERRDATE$] HG error: (glob) $LOCALIP - - [$ERRDATE$] HG error: (glob)
$LOCALIP - - [$ERRDATE$] Exception happened during processing request '/.hg/lfs/objects/276f73cfd75f9fb519810df5f5d96d6594ca2521abd86cbcd92122f7d51a1f3d': (glob) $LOCALIP - - [$ERRDATE$] Exception happened during processing request '/.hg/lfs/objects/276f73cfd75f9fb519810df5f5d96d6594ca2521abd86cbcd92122f7d51a1f3d': (glob)
Traceback (most recent call last): Traceback (most recent call last):
self.do_write() self.do_write()
self.do_hgweb() self.do_hgweb()