dulwich/client.py

   1 # client.py -- Implementation of the server side git protocols
   2 # Copyright (C) 2008-2013 Jelmer Vernooij <jelmer@samba.org>
   3 # Copyright (C) 2008 John Carr
   4 #
   5 # This program is free software; you can redistribute it and/or
   6 # modify it under the terms of the GNU General Public License
   7 # as published by the Free Software Foundation; either version 2
   8 # or (at your option) a later version of the License.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program; if not, write to the Free Software
  17 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  18 # MA  02110-1301, USA.
  19
  20 """Client side support for the Git protocol.
  21
  22 The Dulwich client supports the following capabilities:
  23
  24  * thin-pack
  25  * multi_ack_detailed
  26  * multi_ack
  27  * side-band-64k
  28  * ofs-delta
  29  * report-status
  30  * delete-refs
  31
  32 Known capabilities that are not supported:
  33
  34  * shallow
  35  * no-progress
  36  * include-tag
  37 """
  38
  39 __docformat__ = 'restructuredText'
  40
  41 from cStringIO import StringIO
  42 import select
  43 import socket
  44 import subprocess
  45 import urllib2
  46 import urlparse
  47
  48 from dulwich.errors import (
  49     GitProtocolError,
  50     NotGitRepository,
  51     SendPackError,
  52     UpdateRefsError,
  53     )
  54 from dulwich.protocol import (
  55     _RBUFSIZE,
  56     PktLineParser,
  57     Protocol,
  58     ProtocolFile,
  59     TCP_GIT_PORT,
  60     ZERO_SHA,
  61     extract_capabilities,
  62     )
  63 from dulwich.pack import (
  64     write_pack_objects,
  65     )
  66 from dulwich.refs import (
  67     read_info_refs,
  68     )
  69
  70
  71 # Python 2.6.6 included these in urlparse.uses_netloc upstream. Do
  72 # monkeypatching to enable similar behaviour in earlier Pythons:
  73 for scheme in ('git', 'git+ssh'):
  74     if scheme not in urlparse.uses_netloc:
  75         urlparse.uses_netloc.append(scheme)
  76
  77 def _fileno_can_read(fileno):
  78     """Check if a file descriptor is readable."""
  79     return len(select.select([fileno], [], [], 0)[0]) > 0
  80
  81 COMMON_CAPABILITIES = ['ofs-delta', 'side-band-64k']
  82 FETCH_CAPABILITIES = ['thin-pack', 'multi_ack', 'multi_ack_detailed'] + COMMON_CAPABILITIES
  83 SEND_CAPABILITIES = ['report-status'] + COMMON_CAPABILITIES
  84
  85
  86 class ReportStatusParser(object):
  87     """Handle status as reported by servers with the 'report-status' capability.
  88     """
  89
  90     def __init__(self):
  91         self._done = False
  92         self._pack_status = None
  93         self._ref_status_ok = True
  94         self._ref_statuses = []
  95
  96     def check(self):
  97         """Check if there were any errors and, if so, raise exceptions.
  98
  99         :raise SendPackError: Raised when the server could not unpack
 100         :raise UpdateRefsError: Raised when refs could not be updated
 101         """
 102         if self._pack_status not in ('unpack ok', None):
 103             raise SendPackError(self._pack_status)
 104         if not self._ref_status_ok:
 105             ref_status = {}
 106             ok = set()
 107             for status in self._ref_statuses:
 108                 if ' ' not in status:
 109                     # malformed response, move on to the next one
 110                     continue
 111                 status, ref = status.split(' ', 1)
 112
 113                 if status == 'ng':
 114                     if ' ' in ref:
 115                         ref, status = ref.split(' ', 1)
 116                 else:
 117                     ok.add(ref)
 118                 ref_status[ref] = status
 119             raise UpdateRefsError('%s failed to update' %
 120                                   ', '.join([ref for ref in ref_status
 121                                              if ref not in ok]),
 122                                   ref_status=ref_status)
 123
 124     def handle_packet(self, pkt):
 125         """Handle a packet.
 126
 127         :raise GitProtocolError: Raised when packets are received after a
 128             flush packet.
 129         """
 130         if self._done:
 131             raise GitProtocolError("received more data after status report")
 132         if pkt is None:
 133             self._done = True
 134             return
 135         if self._pack_status is None:
 136             self._pack_status = pkt.strip()
 137         else:
 138             ref_status = pkt.strip()
 139             self._ref_statuses.append(ref_status)
 140             if not ref_status.startswith('ok '):
 141                 self._ref_status_ok = False
 142
 143
 144 def read_pkt_refs(proto):
 145     server_capabilities = None
 146     refs = {}
 147     # Receive refs from server
 148     for pkt in proto.read_pkt_seq():
 149         (sha, ref) = pkt.rstrip('\n').split(None, 1)
 150         if sha == 'ERR':
 151             raise GitProtocolError(ref)
 152         if server_capabilities is None:
 153             (ref, server_capabilities) = extract_capabilities(ref)
 154         refs[ref] = sha
 155
 156     if len(refs) == 0:
 157         return None, set([])
 158     return refs, set(server_capabilities)
 159
 160
 161 # TODO(durin42): this doesn't correctly degrade if the server doesn't
 162 # support some capabilities. This should work properly with servers
 163 # that don't support multi_ack.
 164 class GitClient(object):
 165     """Git smart server client.
 166
 167     """
 168
 169     def __init__(self, thin_packs=True, report_activity=None):
 170         """Create a new GitClient instance.
 171
 172         :param thin_packs: Whether or not thin packs should be retrieved
 173         :param report_activity: Optional callback for reporting transport
 174             activity.
 175         """
 176         self._report_activity = report_activity
 177         self._report_status_parser = None
 178         self._fetch_capabilities = set(FETCH_CAPABILITIES)
 179         self._send_capabilities = set(SEND_CAPABILITIES)
 180         if not thin_packs:
 181             self._fetch_capabilities.remove('thin-pack')
 182
 183     def send_pack(self, path, determine_wants, generate_pack_contents,
 184                   progress=None):
 185         """Upload a pack to a remote repository.
 186
 187         :param path: Repository path
 188         :param generate_pack_contents: Function that can return a sequence of the
 189             shas of the objects to upload.
 190         :param progress: Optional progress function
 191
 192         :raises SendPackError: if server rejects the pack data
 193         :raises UpdateRefsError: if the server supports report-status
 194                                  and rejects ref updates
 195         """
 196         raise NotImplementedError(self.send_pack)
 197
 198     def fetch(self, path, target, determine_wants=None, progress=None):
 199         """Fetch into a target repository.
 200
 201         :param path: Path to fetch from
 202         :param target: Target repository to fetch into
 203         :param determine_wants: Optional function to determine what refs
 204             to fetch
 205         :param progress: Optional progress function
 206         :return: remote refs as dictionary
 207         """
 208         if determine_wants is None:
 209             determine_wants = target.object_store.determine_wants_all
 210         f, commit, abort = target.object_store.add_pack()
 211         try:
 212             result = self.fetch_pack(path, determine_wants,
 213                     target.get_graph_walker(), f.write, progress)
 214         except:
 215             abort()
 216             raise
 217         else:
 218             commit()
 219         return result
 220
 221     def fetch_pack(self, path, determine_wants, graph_walker, pack_data,
 222                    progress=None):
 223         """Retrieve a pack from a git smart server.
 224
 225         :param determine_wants: Callback that returns list of commits to fetch
 226         :param graph_walker: Object with next() and ack().
 227         :param pack_data: Callback called for each bit of data in the pack
 228         :param progress: Callback for progress reports (strings)
 229         """
 230         raise NotImplementedError(self.fetch_pack)
 231
 232     def _parse_status_report(self, proto):
 233         unpack = proto.read_pkt_line().strip()
 234         if unpack != 'unpack ok':
 235             st = True
 236             # flush remaining error data
 237             while st is not None:
 238                 st = proto.read_pkt_line()
 239             raise SendPackError(unpack)
 240         statuses = []
 241         errs = False
 242         ref_status = proto.read_pkt_line()
 243         while ref_status:
 244             ref_status = ref_status.strip()
 245             statuses.append(ref_status)
 246             if not ref_status.startswith('ok '):
 247                 errs = True
 248             ref_status = proto.read_pkt_line()
 249
 250         if errs:
 251             ref_status = {}
 252             ok = set()
 253             for status in statuses:
 254                 if ' ' not in status:
 255                     # malformed response, move on to the next one
 256                     continue
 257                 status, ref = status.split(' ', 1)
 258
 259                 if status == 'ng':
 260                     if ' ' in ref:
 261                         ref, status = ref.split(' ', 1)
 262                 else:
 263                     ok.add(ref)
 264                 ref_status[ref] = status
 265             raise UpdateRefsError('%s failed to update' %
 266                                   ', '.join([ref for ref in ref_status
 267                                              if ref not in ok]),
 268                                   ref_status=ref_status)
 269
 270     def _read_side_band64k_data(self, proto, channel_callbacks):
 271         """Read per-channel data.
 272
 273         This requires the side-band-64k capability.
 274
 275         :param proto: Protocol object to read from
 276         :param channel_callbacks: Dictionary mapping channels to packet
 277             handlers to use. None for a callback discards channel data.
 278         """
 279         for pkt in proto.read_pkt_seq():
 280             channel = ord(pkt[0])
 281             pkt = pkt[1:]
 282             try:
 283                 cb = channel_callbacks[channel]
 284             except KeyError:
 285                 raise AssertionError('Invalid sideband channel %d' % channel)
 286             else:
 287                 if cb is not None:
 288                     cb(pkt)
 289
 290     def _handle_receive_pack_head(self, proto, capabilities, old_refs, new_refs):
 291         """Handle the head of a 'git-receive-pack' request.
 292
 293         :param proto: Protocol object to read from
 294         :param capabilities: List of negotiated capabilities
 295         :param old_refs: Old refs, as received from the server
 296         :param new_refs: New refs
 297         :return: (have, want) tuple
 298         """
 299         want = []
 300         have = [x for x in old_refs.values() if not x == ZERO_SHA]
 301         sent_capabilities = False
 302
 303         for refname in set(new_refs.keys() + old_refs.keys()):
 304             old_sha1 = old_refs.get(refname, ZERO_SHA)
 305             new_sha1 = new_refs.get(refname, ZERO_SHA)
 306
 307             if old_sha1 != new_sha1:
 308                 if sent_capabilities:
 309                     proto.write_pkt_line('%s %s %s' % (old_sha1, new_sha1,
 310                                                             refname))
 311                 else:
 312                     proto.write_pkt_line(
 313                       '%s %s %s\0%s' % (old_sha1, new_sha1, refname,
 314                                         ' '.join(capabilities)))
 315                     sent_capabilities = True
 316             if new_sha1 not in have and new_sha1 != ZERO_SHA:
 317                 want.append(new_sha1)
 318         proto.write_pkt_line(None)
 319         return (have, want)
 320
 321     def _handle_receive_pack_tail(self, proto, capabilities, progress=None):
 322         """Handle the tail of a 'git-receive-pack' request.
 323
 324         :param proto: Protocol object to read from
 325         :param capabilities: List of negotiated capabilities
 326         :param progress: Optional progress reporting function
 327         """
 328         if "side-band-64k" in capabilities:
 329             if progress is None:
 330                 progress = lambda x: None
 331             channel_callbacks = { 2: progress }
 332             if 'report-status' in capabilities:
 333                 channel_callbacks[1] = PktLineParser(
 334                     self._report_status_parser.handle_packet).parse
 335             self._read_side_band64k_data(proto, channel_callbacks)
 336         else:
 337             if 'report-status' in capabilities:
 338                 for pkt in proto.read_pkt_seq():
 339                     self._report_status_parser.handle_packet(pkt)
 340         if self._report_status_parser is not None:
 341             self._report_status_parser.check()
 342         # wait for EOF before returning
 343         data = proto.read()
 344         if data:
 345             raise SendPackError('Unexpected response %r' % data)
 346
 347     def _handle_upload_pack_head(self, proto, capabilities, graph_walker,
 348                                  wants, can_read):
 349         """Handle the head of a 'git-upload-pack' request.
 350
 351         :param proto: Protocol object to read from
 352         :param capabilities: List of negotiated capabilities
 353         :param graph_walker: GraphWalker instance to call .ack() on
 354         :param wants: List of commits to fetch
 355         :param can_read: function that returns a boolean that indicates
 356             whether there is extra graph data to read on proto
 357         """
 358         assert isinstance(wants, list) and type(wants[0]) == str
 359         proto.write_pkt_line('want %s %s\n' % (
 360             wants[0], ' '.join(capabilities)))
 361         for want in wants[1:]:
 362             proto.write_pkt_line('want %s\n' % want)
 363         proto.write_pkt_line(None)
 364         have = graph_walker.next()
 365         while have:
 366             proto.write_pkt_line('have %s\n' % have)
 367             if can_read():
 368                 pkt = proto.read_pkt_line()
 369                 parts = pkt.rstrip('\n').split(' ')
 370                 if parts[0] == 'ACK':
 371                     graph_walker.ack(parts[1])
 372                     if parts[2] in ('continue', 'common'):
 373                         pass
 374                     elif parts[2] == 'ready':
 375                         break
 376                     else:
 377                         raise AssertionError(
 378                             "%s not in ('continue', 'ready', 'common)" %
 379                             parts[2])
 380             have = graph_walker.next()
 381         proto.write_pkt_line('done\n')
 382
 383     def _handle_upload_pack_tail(self, proto, capabilities, graph_walker,
 384                                  pack_data, progress=None, rbufsize=_RBUFSIZE):
 385         """Handle the tail of a 'git-upload-pack' request.
 386
 387         :param proto: Protocol object to read from
 388         :param capabilities: List of negotiated capabilities
 389         :param graph_walker: GraphWalker instance to call .ack() on
 390         :param pack_data: Function to call with pack data
 391         :param progress: Optional progress reporting function
 392         :param rbufsize: Read buffer size
 393         """
 394         pkt = proto.read_pkt_line()
 395         while pkt:
 396             parts = pkt.rstrip('\n').split(' ')
 397             if parts[0] == 'ACK':
 398                 graph_walker.ack(pkt.split(' ')[1])
 399             if len(parts) < 3 or parts[2] not in (
 400                     'ready', 'continue', 'common'):
 401                 break
 402             pkt = proto.read_pkt_line()
 403         if "side-band-64k" in capabilities:
 404             if progress is None:
 405                 # Just ignore progress data
 406                 progress = lambda x: None
 407             self._read_side_band64k_data(proto, {1: pack_data, 2: progress})
 408             # wait for EOF before returning
 409             data = proto.read()
 410             if data:
 411                 raise Exception('Unexpected response %r' % data)
 412         else:
 413             while True:
 414                 data = proto.read(rbufsize)
 415                 if data == "":
 416                     break
 417                 pack_data(data)
 418
 419
 420 class TraditionalGitClient(GitClient):
 421     """Traditional Git client."""
 422
 423     def _connect(self, cmd, path):
 424         """Create a connection to the server.
 425
 426         This method is abstract - concrete implementations should
 427         implement their own variant which connects to the server and
 428         returns an initialized Protocol object with the service ready
 429         for use and a can_read function which may be used to see if
 430         reads would block.
 431
 432         :param cmd: The git service name to which we should connect.
 433         :param path: The path we should pass to the service.
 434         """
 435         raise NotImplementedError()
 436
 437     def send_pack(self, path, determine_wants, generate_pack_contents,
 438                   progress=None):
 439         """Upload a pack to a remote repository.
 440
 441         :param path: Repository path
 442         :param generate_pack_contents: Function that can return a sequence of the
 443             shas of the objects to upload.
 444         :param progress: Optional callback called with progress updates
 445
 446         :raises SendPackError: if server rejects the pack data
 447         :raises UpdateRefsError: if the server supports report-status
 448                                  and rejects ref updates
 449         """
 450         proto, unused_can_read = self._connect('receive-pack', path)
 451         old_refs, server_capabilities = read_pkt_refs(proto)
 452         negotiated_capabilities = self._send_capabilities & server_capabilities
 453
 454         if 'report-status' in negotiated_capabilities:
 455             self._report_status_parser = ReportStatusParser()
 456         report_status_parser = self._report_status_parser
 457
 458         try:
 459             new_refs = orig_new_refs = determine_wants(dict(old_refs))
 460         except:
 461             proto.write_pkt_line(None)
 462             raise
 463
 464         if not 'delete-refs' in server_capabilities:
 465             # Server does not support deletions. Fail later.
 466             def remove_del(pair):
 467                 if pair[1] == ZERO_SHA:
 468                     if 'report-status' in negotiated_capabilities:
 469                         report_status_parser._ref_statuses.append(
 470                             'ng %s remote does not support deleting refs'
 471                             % pair[1])
 472                         report_status_parser._ref_status_ok = False
 473                     return False
 474                 else:
 475                     return True
 476
 477             new_refs = dict(
 478                 filter(
 479                     remove_del,
 480                     [(ref, sha) for ref, sha in new_refs.iteritems()]))
 481
 482         if new_refs is None:
 483             proto.write_pkt_line(None)
 484             return old_refs
 485
 486         if len(new_refs) == 0 and len(orig_new_refs):
 487             # NOOP - Original new refs filtered out by policy
 488             proto.write_pkt_line(None)
 489             if self._report_status_parser is not None:
 490                 self._report_status_parser.check()
 491             return old_refs
 492
 493         (have, want) = self._handle_receive_pack_head(proto,
 494             negotiated_capabilities, old_refs, new_refs)
 495         if not want and old_refs == new_refs:
 496             return new_refs
 497         objects = generate_pack_contents(have, want)
 498         if len(objects) > 0:
 499             entries, sha = write_pack_objects(proto.write_file(), objects)
 500         elif len(set(new_refs.values()) - set([ZERO_SHA])) > 0:
 501             # Check for valid create/update refs
 502             filtered_new_refs = \
 503                 dict([(ref, sha) for ref, sha in new_refs.iteritems()
 504                      if sha != ZERO_SHA])
 505             if len(set(filtered_new_refs.iteritems()) -
 506                     set(old_refs.iteritems())) > 0:
 507                 entries, sha = write_pack_objects(proto.write_file(), objects)
 508
 509         self._handle_receive_pack_tail(proto, negotiated_capabilities,
 510             progress)
 511         return new_refs
 512
 513     def fetch_pack(self, path, determine_wants, graph_walker, pack_data,
 514                    progress=None):
 515         """Retrieve a pack from a git smart server.
 516
 517         :param determine_wants: Callback that returns list of commits to fetch
 518         :param graph_walker: Object with next() and ack().
 519         :param pack_data: Callback called for each bit of data in the pack
 520         :param progress: Callback for progress reports (strings)
 521         """
 522         proto, can_read = self._connect('upload-pack', path)
 523         refs, server_capabilities = read_pkt_refs(proto)
 524         negotiated_capabilities = self._fetch_capabilities & server_capabilities
 525
 526         if refs is None:
 527             proto.write_pkt_line(None)
 528             return refs
 529
 530         try:
 531             wants = determine_wants(refs)
 532         except:
 533             proto.write_pkt_line(None)
 534             raise
 535         if wants is not None:
 536             wants = [cid for cid in wants if cid != ZERO_SHA]
 537         if not wants:
 538             proto.write_pkt_line(None)
 539             return refs
 540         self._handle_upload_pack_head(proto, negotiated_capabilities,
 541             graph_walker, wants, can_read)
 542         self._handle_upload_pack_tail(proto, negotiated_capabilities,
 543             graph_walker, pack_data, progress)
 544         return refs
 545
 546     def archive(self, path, committish, write_data, progress=None):
 547         proto, can_read = self._connect('upload-archive', path)
 548         proto.write_pkt_line("argument %s" % committish)
 549         proto.write_pkt_line(None)
 550         pkt = proto.read_pkt_line()
 551         if pkt == "NACK\n":
 552             return
 553         elif pkt == "ACK\n":
 554             pass
 555         elif pkt.startswith("ERR "):
 556             raise GitProtocolError(pkt[4:].rstrip("\n"))
 557         else:
 558             raise AssertionError("invalid response %r" % pkt)
 559         ret = proto.read_pkt_line()
 560         if ret is not None:
 561             raise AssertionError("expected pkt tail")
 562         self._read_side_band64k_data(proto, {1: write_data, 2: progress})
 563
 564
 565 class TCPGitClient(TraditionalGitClient):
 566     """A Git Client that works over TCP directly (i.e. git://)."""
 567
 568     def __init__(self, host, port=None, *args, **kwargs):
 569         if port is None:
 570             port = TCP_GIT_PORT
 571         self._host = host
 572         self._port = port
 573         TraditionalGitClient.__init__(self, *args, **kwargs)
 574
 575     def _connect(self, cmd, path):
 576         sockaddrs = socket.getaddrinfo(self._host, self._port,
 577             socket.AF_UNSPEC, socket.SOCK_STREAM)
 578         s = None
 579         err = socket.error("no address found for %s" % self._host)
 580         for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
 581             s = socket.socket(family, socktype, proto)
 582             s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 583             try:
 584                 s.connect(sockaddr)
 585                 break
 586             except socket.error as err:
 587                 if s is not None:
 588                     s.close()
 589                 s = None
 590         if s is None:
 591             raise err
 592         # -1 means system default buffering
 593         rfile = s.makefile('rb', -1)
 594         # 0 means unbuffered
 595         wfile = s.makefile('wb', 0)
 596         proto = Protocol(rfile.read, wfile.write,
 597                          report_activity=self._report_activity)
 598         if path.startswith("/~"):
 599             path = path[1:]
 600         proto.send_cmd('git-%s' % cmd, path, 'host=%s' % self._host)
 601         return proto, lambda: _fileno_can_read(s)
 602
 603
 604 class SubprocessWrapper(object):
 605     """A socket-like object that talks to a subprocess via pipes."""
 606
 607     def __init__(self, proc):
 608         self.proc = proc
 609         self.read = proc.stdout.read
 610         self.write = proc.stdin.write
 611
 612     def can_read(self):
 613         if subprocess.mswindows:
 614             from msvcrt import get_osfhandle
 615             from win32pipe import PeekNamedPipe
 616             handle = get_osfhandle(self.proc.stdout.fileno())
 617             return PeekNamedPipe(handle, 0)[2] != 0
 618         else:
 619             return _fileno_can_read(self.proc.stdout.fileno())
 620
 621     def close(self):
 622         self.proc.stdin.close()
 623         self.proc.stdout.close()
 624         self.proc.wait()
 625
 626
 627 class SubprocessGitClient(TraditionalGitClient):
 628     """Git client that talks to a server using a subprocess."""
 629
 630     def __init__(self, *args, **kwargs):
 631         self._connection = None
 632         self._stderr = None
 633         self._stderr = kwargs.get('stderr')
 634         if 'stderr' in kwargs:
 635             del kwargs['stderr']
 636         TraditionalGitClient.__init__(self, *args, **kwargs)
 637
 638     def _connect(self, service, path):
 639         import subprocess
 640         argv = ['git', service, path]
 641         p = SubprocessWrapper(
 642             subprocess.Popen(argv, bufsize=0, stdin=subprocess.PIPE,
 643                              stdout=subprocess.PIPE,
 644                              stderr=self._stderr))
 645         return Protocol(p.read, p.write,
 646                         report_activity=self._report_activity), p.can_read
 647
 648
 649 class LocalGitClient(GitClient):
 650     """Git Client that just uses a local Repo."""
 651
 652     def __init__(self, thin_packs=True, report_activity=None):
 653         """Create a new LocalGitClient instance.
 654
 655         :param path: Path to the local repository
 656         :param thin_packs: Whether or not thin packs should be retrieved
 657         :param report_activity: Optional callback for reporting transport
 658             activity.
 659         """
 660         self._report_activity = report_activity
 661         # Ignore the thin_packs argument
 662
 663     def send_pack(self, path, determine_wants, generate_pack_contents,
 664                   progress=None):
 665         """Upload a pack to a remote repository.
 666
 667         :param path: Repository path
 668         :param generate_pack_contents: Function that can return a sequence of the
 669             shas of the objects to upload.
 670         :param progress: Optional progress function
 671
 672         :raises SendPackError: if server rejects the pack data
 673         :raises UpdateRefsError: if the server supports report-status
 674                                  and rejects ref updates
 675         """
 676         raise NotImplementedError(self.send_pack)
 677
 678     def fetch(self, path, target, determine_wants=None, progress=None):
 679         """Fetch into a target repository.
 680
 681         :param path: Path to fetch from
 682         :param target: Target repository to fetch into
 683         :param determine_wants: Optional function to determine what refs
 684             to fetch
 685         :param progress: Optional progress function
 686         :return: remote refs as dictionary
 687         """
 688         from dulwich.repo import Repo
 689         r = Repo(path)
 690         return r.fetch(target, determine_wants=determine_wants, progress=progress)
 691
 692     def fetch_pack(self, path, determine_wants, graph_walker, pack_data,
 693                    progress=None):
 694         """Retrieve a pack from a git smart server.
 695
 696         :param determine_wants: Callback that returns list of commits to fetch
 697         :param graph_walker: Object with next() and ack().
 698         :param pack_data: Callback called for each bit of data in the pack
 699         :param progress: Callback for progress reports (strings)
 700         """
 701         from dulwich.repo import Repo
 702         r = Repo(path)
 703         objects_iter = r.fetch_objects(determine_wants, graph_walker, progress)
 704
 705         # Did the process short-circuit (e.g. in a stateless RPC call)? Note
 706         # that the client still expects a 0-object pack in most cases.
 707         if objects_iter is None:
 708             return
 709         write_pack_objects(ProtocolFile(None, pack_data), objects_iter)
 710
 711
 712 # What Git client to use for local access
 713 default_local_git_client_cls = SubprocessGitClient
 714
 715 class SSHVendor(object):
 716     """A client side SSH implementation."""
 717
 718     def connect_ssh(self, host, command, username=None, port=None):
 719         import warnings
 720         warnings.warn(
 721             "SSHVendor.connect_ssh has been renamed to SSHVendor.run_command",
 722             DeprecationWarning)
 723         return self.run_command(host, command, username=username, port=port)
 724
 725     def run_command(self, host, command, username=None, port=None):
 726         """Connect to an SSH server.
 727
 728         Run a command remotely and return a file-like object for interaction
 729         with the remote command.
 730
 731         :param host: Host name
 732         :param command: Command to run
 733         :param username: Optional ame of user to log in as
 734         :param port: Optional SSH port to use
 735         """
 736         raise NotImplementedError(self.run_command)
 737
 738
 739 class SubprocessSSHVendor(SSHVendor):
 740     """SSH vendor that shells out to the local 'ssh' command."""
 741
 742     def run_command(self, host, command, username=None, port=None):
 743         import subprocess
 744         #FIXME: This has no way to deal with passwords..
 745         args = ['ssh', '-x']
 746         if port is not None:
 747             args.extend(['-p', str(port)])
 748         if username is not None:
 749             host = '%s@%s' % (username, host)
 750         args.append(host)
 751         proc = subprocess.Popen(args + command,
 752                                 stdin=subprocess.PIPE,
 753                                 stdout=subprocess.PIPE)
 754         return SubprocessWrapper(proc)
 755
 756
 757 try:
 758     import paramiko
 759 except ImportError:
 760     pass
 761 else:
 762     import threading
 763
 764     class ParamikoWrapper(object):
 765         STDERR_READ_N = 2048  # 2k
 766
 767         def __init__(self, client, channel, progress_stderr=None):
 768             self.client = client
 769             self.channel = channel
 770             self.progress_stderr = progress_stderr
 771             self.should_monitor = bool(progress_stderr) or True
 772             self.monitor_thread = None
 773             self.stderr = ''
 774
 775             # Channel must block
 776             self.channel.setblocking(True)
 777
 778             # Start
 779             if self.should_monitor:
 780                 self.monitor_thread = threading.Thread(target=self.monitor_stderr)
 781                 self.monitor_thread.start()
 782
 783         def monitor_stderr(self):
 784             while self.should_monitor:
 785                 # Block and read
 786                 data = self.read_stderr(self.STDERR_READ_N)
 787
 788                 # Socket closed
 789                 if not data:
 790                     self.should_monitor = False
 791                     break
 792
 793                 # Emit data
 794                 if self.progress_stderr:
 795                     self.progress_stderr(data)
 796
 797                 # Append to buffer
 798                 self.stderr += data
 799
 800         def stop_monitoring(self):
 801             # Stop StdErr thread
 802             if self.should_monitor:
 803                 self.should_monitor = False
 804                 self.monitor_thread.join()
 805
 806                 # Get left over data
 807                 data = self.channel.in_stderr_buffer.empty()
 808                 self.stderr += data
 809
 810         def can_read(self):
 811             return self.channel.recv_ready()
 812
 813         def write(self, data):
 814             return self.channel.sendall(data)
 815
 816         def read_stderr(self, n):
 817             return self.channel.recv_stderr(n)
 818
 819         def read(self, n=None):
 820             data = self.channel.recv(n)
 821             data_len = len(data)
 822
 823             # Closed socket
 824             if not data:
 825                 return
 826
 827             # Read more if needed
 828             if n and data_len < n:
 829                 diff_len = n - data_len
 830                 return data + self.read(diff_len)
 831             return data
 832
 833         def close(self):
 834             self.channel.close()
 835             self.stop_monitoring()
 836
 837         def __del__(self):
 838             self.close()
 839
 840     class ParamikoSSHVendor(object):
 841
 842         def __init__(self):
 843             self.ssh_kwargs = {}
 844
 845         def run_command(self, host, command, username=None, port=None,
 846                 progress_stderr=None):
 847
 848             # Paramiko needs an explicit port. None is not valid
 849             if port is None:
 850                 port = 22
 851
 852             client = paramiko.SSHClient()
 853
 854             policy = paramiko.client.MissingHostKeyPolicy()
 855             client.set_missing_host_key_policy(policy)
 856             client.connect(host, username=username, port=port,
 857                            **self.ssh_kwargs)
 858
 859             # Open SSH session
 860             channel = client.get_transport().open_session()
 861
 862             # Run commands
 863             apply(channel.exec_command, command)
 864
 865             return ParamikoWrapper(client, channel,
 866                     progress_stderr=progress_stderr)
 867
 868
 869 # Can be overridden by users
 870 get_ssh_vendor = SubprocessSSHVendor
 871
 872
 873 class SSHGitClient(TraditionalGitClient):
 874
 875     def __init__(self, host, port=None, username=None, *args, **kwargs):
 876         self.host = host
 877         self.port = port
 878         self.username = username
 879         TraditionalGitClient.__init__(self, *args, **kwargs)
 880         self.alternative_paths = {}
 881
 882     def _get_cmd_path(self, cmd):
 883         return self.alternative_paths.get(cmd, 'git-%s' % cmd)
 884
 885     def _connect(self, cmd, path):
 886         if path.startswith("/~"):
 887             path = path[1:]
 888         con = get_ssh_vendor().run_command(
 889             self.host, ["%s '%s'" % (self._get_cmd_path(cmd), path)],
 890             port=self.port, username=self.username)
 891         return (Protocol(con.read, con.write, report_activity=self._report_activity),
 892                 con.can_read)
 893
 894
 895 class HttpGitClient(GitClient):
 896
 897     def __init__(self, base_url, dumb=None, handlers=None, *args, **kwargs):
 898         self.base_url = base_url.rstrip("/") + "/"
 899         self.dumb = dumb
 900         self.handlers = handlers or []
 901         GitClient.__init__(self, *args, **kwargs)
 902
 903     def _get_url(self, path):
 904         return urlparse.urljoin(self.base_url, path).rstrip("/") + "/"
 905
 906     def _http_request(self, url, headers={}, data=None):
 907         opener = urllib2.build_opener(*self.handlers)
 908         urllib2.install_opener(opener)
 909         req = urllib2.Request(url, headers=headers, data=data)
 910         try:
 911             resp = self._perform(req)
 912         except urllib2.HTTPError as e:
 913             if e.code == 404:
 914                 raise NotGitRepository()
 915             if e.code != 200:
 916                 raise GitProtocolError("unexpected http response %d" % e.code)
 917         return resp
 918
 919     def _perform(self, req):
 920         """Perform a HTTP request.
 921
 922         This is provided so subclasses can provide their own version.
 923
 924         :param req: urllib2.Request instance
 925         :return: matching response
 926         """
 927         return urllib2.urlopen(req)
 928
 929     def _discover_references(self, service, url):
 930         assert url[-1] == "/"
 931         url = urlparse.urljoin(url, "info/refs")
 932         headers = {}
 933         if self.dumb != False:
 934             url += "?service=%s" % service
 935             headers["Content-Type"] = "application/x-%s-request" % service
 936         resp = self._http_request(url, headers)
 937         self.dumb = (not resp.info().gettype().startswith("application/x-git-"))
 938         if not self.dumb:
 939             proto = Protocol(resp.read, None)
 940             # The first line should mention the service
 941             pkts = list(proto.read_pkt_seq())
 942             if pkts != [('# service=%s\n' % service)]:
 943                 raise GitProtocolError(
 944                     "unexpected first line %r from smart server" % pkts)
 945             return read_pkt_refs(proto)
 946         else:
 947             return read_info_refs(resp), set()
 948
 949     def _smart_request(self, service, url, data):
 950         assert url[-1] == "/"
 951         url = urlparse.urljoin(url, service)
 952         headers = {"Content-Type": "application/x-%s-request" % service}
 953         resp = self._http_request(url, headers, data)
 954         if resp.info().gettype() != ("application/x-%s-result" % service):
 955             raise GitProtocolError("Invalid content-type from server: %s"
 956                 % resp.info().gettype())
 957         return resp
 958
 959     def send_pack(self, path, determine_wants, generate_pack_contents,
 960                   progress=None):
 961         """Upload a pack to a remote repository.
 962
 963         :param path: Repository path
 964         :param generate_pack_contents: Function that can return a sequence of the
 965             shas of the objects to upload.
 966         :param progress: Optional progress function
 967
 968         :raises SendPackError: if server rejects the pack data
 969         :raises UpdateRefsError: if the server supports report-status
 970                                  and rejects ref updates
 971         """
 972         url = self._get_url(path)
 973         old_refs, server_capabilities = self._discover_references(
 974             "git-receive-pack", url)
 975         negotiated_capabilities = self._send_capabilities & server_capabilities
 976
 977         if 'report-status' in negotiated_capabilities:
 978             self._report_status_parser = ReportStatusParser()
 979
 980         new_refs = determine_wants(dict(old_refs))
 981         if new_refs is None:
 982             return old_refs
 983         if self.dumb:
 984             raise NotImplementedError(self.fetch_pack)
 985         req_data = StringIO()
 986         req_proto = Protocol(None, req_data.write)
 987         (have, want) = self._handle_receive_pack_head(
 988             req_proto, negotiated_capabilities, old_refs, new_refs)
 989         if not want and old_refs == new_refs:
 990             return new_refs
 991         objects = generate_pack_contents(have, want)
 992         if len(objects) > 0:
 993             entries, sha = write_pack_objects(req_proto.write_file(), objects)
 994         resp = self._smart_request("git-receive-pack", url,
 995             data=req_data.getvalue())
 996         resp_proto = Protocol(resp.read, None)
 997         self._handle_receive_pack_tail(resp_proto, negotiated_capabilities,
 998             progress)
 999         return new_refs
1000
1001     def fetch_pack(self, path, determine_wants, graph_walker, pack_data,
1002                    progress=None):
1003         """Retrieve a pack from a git smart server.
1004
1005         :param determine_wants: Callback that returns list of commits to fetch
1006         :param graph_walker: Object with next() and ack().
1007         :param pack_data: Callback called for each bit of data in the pack
1008         :param progress: Callback for progress reports (strings)
1009         :return: Dictionary with the refs of the remote repository
1010         """
1011         url = self._get_url(path)
1012         refs, server_capabilities = self._discover_references(
1013             "git-upload-pack", url)
1014         negotiated_capabilities = self._fetch_capabilities & server_capabilities
1015         wants = determine_wants(refs)
1016         if wants is not None:
1017             wants = [cid for cid in wants if cid != ZERO_SHA]
1018         if not wants:
1019             return refs
1020         if self.dumb:
1021             raise NotImplementedError(self.send_pack)
1022         req_data = StringIO()
1023         req_proto = Protocol(None, req_data.write)
1024         self._handle_upload_pack_head(req_proto,
1025             negotiated_capabilities, graph_walker, wants,
1026             lambda: False)
1027         resp = self._smart_request("git-upload-pack", url,
1028             data=req_data.getvalue())
1029         resp_proto = Protocol(resp.read, None)
1030         self._handle_upload_pack_tail(resp_proto, negotiated_capabilities,
1031             graph_walker, pack_data, progress)
1032         return refs
1033
1034
1035 def get_transport_and_path_from_url(url, **kwargs):
1036     """Obtain a git client from a URL.
1037
1038     :param url: URL to open
1039     :param thin_packs: Whether or not thin packs should be retrieved
1040     :param report_activity: Optional callback for reporting transport
1041         activity.
1042     :return: Tuple with client instance and relative path.
1043     """
1044     parsed = urlparse.urlparse(url)
1045     if parsed.scheme == 'git':
1046         return (TCPGitClient(parsed.hostname, port=parsed.port, **kwargs),
1047                 parsed.path)
1048     elif parsed.scheme == 'git+ssh':
1049         path = parsed.path
1050         if path.startswith('/'):
1051             path = parsed.path[1:]
1052         return SSHGitClient(parsed.hostname, port=parsed.port,
1053                             username=parsed.username, **kwargs), path
1054     elif parsed.scheme in ('http', 'https'):
1055         return HttpGitClient(urlparse.urlunparse(parsed), **kwargs), parsed.path
1056     elif parsed.scheme == 'file':
1057         return default_local_git_client_cls(**kwargs), parsed.path
1058
1059     raise ValueError("unknown scheme '%s'" % parsed.scheme)
1060
1061
1062 def get_transport_and_path(location, **kwargs):
1063     """Obtain a git client from a URL.
1064
1065     :param location: URL or path
1066     :param thin_packs: Whether or not thin packs should be retrieved
1067     :param report_activity: Optional callback for reporting transport
1068         activity.
1069     :return: Tuple with client instance and relative path.
1070     """
1071     # First, try to parse it as a URL
1072     try:
1073         return get_transport_and_path_from_url(location, **kwargs)
1074     except ValueError:
1075         pass
1076
1077     if ':' in location and not '@' in location:
1078         # SSH with no user@, zero or one leading slash.
1079         (hostname, path) = location.split(':')
1080         return SSHGitClient(hostname, **kwargs), path
1081     elif '@' in location and ':' in location:
1082         # SSH with user@host:foo.
1083         user_host, path = location.split(':')
1084         user, host = user_host.rsplit('@')
1085         return SSHGitClient(host, username=user, **kwargs), path
1086
1087     # Otherwise, assume it's a local path.
1088     return default_local_git_client_cls(**kwargs), location