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