dulwich/client.py

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