Initial import.

.gitignore

@@ -0,0 +1,6 @@
+*.py[co]
+*.egg-info
+/stage
+/.pydoctor.pickle
+/apidocs
+/gitosis/test/tmp

README.rst

@@ -0,0 +1,35 @@
+==========================================================
+ ``gitosis`` -- software for hosting ``git`` repositories
+==========================================================
+
+group -> list of repos
+
+/usr/local/bin/git-shell-enforce-directory
+
+check that the user account (e.g. ``git``) looks valid
+
+ssh keys
+
+regenerate authorized_keys, only touching lines that look safe
+
+allow skipping .git suffix
+
+git-daemon-export-ok
+
+Example configuration::
+
+	[gitosis]
+
+	[group NAME]
+	members = jdoe wsmith @anothergroup
+	writable = foo bar baz/thud
+	readonly = xyzzy
+	map writable visiblename = actualname
+	map readonly visiblename = actualname
+
+	[repo foo]
+	description = blah blah
+	daemon-ok = no
+
+	[gitweb]
+	homelink = http://example.com/

gitosis/__init__.py

@@ -0,0 +1,3 @@
+"""
+gitosis -- software for hosting git repositories
+"""

gitosis/access.py

@@ -0,0 +1,32 @@
+from ConfigParser import NoSectionError, NoOptionError
+
+from gitosis import group
+
+def haveAccess(config, user, mode, path):
+    """
+    Map request for write access to allowed path.
+
+    Note for read-only access, the caller should check for write
+    access too.
+
+    Returns ``None`` for no access, or the physical repository path
+    for access granted to that repository.
+    """
+    for groupname in group.getMembership(config=config, user=user):
+        try:
+            repos = config.get('group %s' % groupname, mode)
+        except (NoSectionError, NoOptionError):
+            repos = []
+        else:
+            repos = repos.split()
+
+        if path in repos:
+            return path
+
+        try:
+            mapping = config.get('group %s' % groupname,
+                                 'map %s %s' % (mode, path))
+        except (NoSectionError, NoOptionError):
+            pass
+        else:
+            return mapping

gitosis/group.py

@@ -0,0 +1,42 @@
+import logging
+from ConfigParser import NoSectionError, NoOptionError
+
+def getMembership(config, user, _seen=None):
+    """
+    Generate groups ``user`` is member of, according to ``config``
+
+    :type config: RawConfigParser
+    :type user: str
+    :param _seen: internal use only
+    """
+    log = logging.getLogger('gitosis.group.getMembership')
+
+    if _seen is None:
+        _seen = set()
+
+    for section in config.sections():
+        GROUP_PREFIX = 'group '
+        if not section.startswith(GROUP_PREFIX):
+            continue
+        group = section[len(GROUP_PREFIX):]
+        if group in _seen:
+            continue
+
+        try:
+            members = config.get(section, 'members')
+        except (NoSectionError, NoOptionError):
+            members = []
+        else:
+            members = members.split()
+
+        if user in members:
+            log.debug('found %(user)r in %(group)r' % dict(
+                user=user,
+                group=group,
+                ))
+            _seen.add(group)
+            yield group
+
+            for member_of in getMembership(config, '@%s' % group,
+                                           _seen=_seen):
+                yield member_of

gitosis/ssh.py

@@ -0,0 +1,104 @@
+import os, errno, re
+
+def readKeys(keydir):
+    """
+    Read SSH public keys from ``keydir/*.pub``
+    """
+    for filename in os.listdir(keydir):
+        if filename.startswith('.'):
+            continue
+        basename, ext = os.path.splitext(filename)
+        if ext != '.pub':
+            continue
+
+        path = os.path.join(keydir, filename)
+        f = file(path)
+        try:
+            line = f.readline()
+        finally:
+            f.close()
+        line = line.rstrip('\n')
+        yield (basename, line)
+
+COMMENT = '### autogenerated by gitosis, DO NOT EDIT'
+
+def generateAuthorizedKeys(keys):
+    TEMPLATE=('command="gitosis-serve %(user)s",no-port-forwarding,'
+              +'no-X11-forwarding,no-agent-forwarding,no-pty %(key)s')
+
+    yield COMMENT
+    for (user, key) in keys:
+        yield TEMPLATE % dict(user=user, key=key)
+
+_COMMAND_RE = re.compile('^command="(/[^ "]+/)?gitosis-serve [^"]+",no-port-forw'
+                         +'arding,no-X11-forwarding,no-agent-forwardi'
+                         +'ng,no-pty .*')
+
+def filterAuthorizedKeys(fp):
+    """
+    Read lines from ``fp``, filter out autogenerated ones.
+
+    Note removes newlines.
+    """
+
+    for line in fp:
+        line = line.rstrip('\n')
+        if line == COMMENT:
+            continue
+        if _COMMAND_RE.match(line):
+            continue
+        yield line
+
+def writeAuthorizedKeys(path, keydir):
+    tmp = '%s.%d.tmp' % (path, os.getpid())
+    try:
+        in_ = file(path)
+    except IOError, e:
+        if e.errno == errno.ENOENT:
+            in_ = None
+        else:
+            raise
+
+    try:
+        out = file(tmp, 'w')
+        try:
+            if in_ is not None:
+                for line in filterAuthorizedKeys(in_):
+                    print >>out, line
+
+            keygen = readKeys(keydir)
+            for line in generateAuthorizedKeys(keygen):
+                print >>out, line
+
+            os.fsync(out)
+        finally:
+            out.close()
+    finally:
+        if in_ is not None:
+            in_.close()
+    os.rename(tmp, path)
+
+def _getParser():
+    import optparse
+    parser = optparse.OptionParser(
+        usage="%prog [--authkeys=FILE] KEYDIR")
+    parser.set_defaults(
+        authkeys=os.path.expanduser('~/.ssh/authorized_keys'),
+        )
+    parser.add_option(
+        "--authkeys",
+        help="path to SSH authorized keys file")
+    return parser
+
+def main():
+    parser = _getParser()
+    (options, args) = parser.parse_args()
+
+    if len(args) != 1:
+        parser.error('Need one argument on the command line.')
+
+    keydir, = args
+
+    writeAuthorizedKeys(
+        path=options.authkeys,
+        keydir=keydir)

gitosis/test/test_access.py

@@ -0,0 +1,79 @@
+from nose.tools import eq_ as eq
+
+from ConfigParser import RawConfigParser
+
+from gitosis import access
+
+def test_write_no_simple():
+    cfg = RawConfigParser()
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='writable', path='foo/bar'),
+       None)
+
+def test_write_yes_simple():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'writable', 'foo/bar')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='writable', path='foo/bar'),
+       'foo/bar')
+
+def test_write_no_simple_wouldHaveReadonly():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'readonly', 'foo/bar')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='writable', path='foo/bar'),
+       None)
+
+def test_write_yes_map():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'map writable foo/bar', 'quux/thud')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='writable', path='foo/bar'),
+       'quux/thud')
+
+def test_write_no_map_wouldHaveReadonly():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'map readonly foo/bar', 'quux/thud')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='writable', path='foo/bar'),
+       None)
+
+def test_read_no_simple():
+    cfg = RawConfigParser()
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='readonly', path='foo/bar'),
+       None)
+
+def test_read_yes_simple():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'readonly', 'foo/bar')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='readonly', path='foo/bar'),
+       'foo/bar')
+
+def test_read_yes_simple_wouldHaveWritable():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'writable', 'foo/bar')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='readonly', path='foo/bar'),
+       None)
+
+def test_read_yes_map():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'map readonly foo/bar', 'quux/thud')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='readonly', path='foo/bar'),
+       'quux/thud')
+
+def test_read_yes_map_wouldHaveWritable():
+    cfg = RawConfigParser()
+    cfg.add_section('group fooers')
+    cfg.set('group fooers', 'members', 'jdoe')
+    cfg.set('group fooers', 'map writable foo/bar', 'quux/thud')
+    eq(access.haveAccess(config=cfg, user='jdoe', mode='readonly', path='foo/bar'),
+       None)

gitosis/test/test_group.py

@@ -0,0 +1,125 @@
+from nose.tools import eq_ as eq, assert_raises
+
+from ConfigParser import RawConfigParser
+
+from gitosis import group
+
+def test_no_emptyConfig():
+    cfg = RawConfigParser()
+    gen = group.getMembership(config=cfg, user='jdoe')
+    assert_raises(StopIteration, gen.next)
+
+def test_no_emptyGroup():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    assert_raises(StopIteration, gen.next)
+
+def test_no_notListed():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'wsmith')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_simple():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'jdoe')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_leading():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'jdoe wsmith')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_trailing():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'wsmith jdoe')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_middle():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'wsmith jdoe danny')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_recurse_one():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'wsmith @smackers')
+    cfg.add_section('group smackers')
+    cfg.set('group smackers', 'members', 'danny jdoe')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'smackers')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_recurse_one_ordering():
+    cfg = RawConfigParser()
+    cfg.add_section('group smackers')
+    cfg.set('group smackers', 'members', 'danny jdoe')
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'wsmith @smackers')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'smackers')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_recurse_three():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', 'wsmith @smackers')
+    cfg.add_section('group smackers')
+    cfg.set('group smackers', 'members', 'danny @snackers')
+    cfg.add_section('group snackers')
+    cfg.set('group snackers', 'members', '@whackers foo')
+    cfg.add_section('group whackers')
+    cfg.set('group whackers', 'members', 'jdoe')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'whackers')
+    eq(gen.next(), 'snackers')
+    eq(gen.next(), 'smackers')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_recurse_junk():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', '@notexist @smackers')
+    cfg.add_section('group smackers')
+    cfg.set('group smackers', 'members', 'jdoe')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'smackers')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_yes_recurse_loop():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', '@smackers')
+    cfg.add_section('group smackers')
+    cfg.set('group smackers', 'members', '@hackers jdoe')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    eq(gen.next(), 'smackers')
+    eq(gen.next(), 'hackers')
+    assert_raises(StopIteration, gen.next)
+
+def test_no_recurse_loop():
+    cfg = RawConfigParser()
+    cfg.add_section('group hackers')
+    cfg.set('group hackers', 'members', '@smackers')
+    cfg.add_section('group smackers')
+    cfg.set('group smackers', 'members', '@hackers')
+    gen = group.getMembership(config=cfg, user='jdoe')
+    assert_raises(StopIteration, gen.next)

gitosis/test/test_ssh.py

@@ -0,0 +1,202 @@
+from nose.tools import eq_ as eq, assert_raises
+
+import os, errno
+from cStringIO import StringIO
+
+from gitosis import ssh
+
+def mkdir(path):
+    try:
+        os.mkdir(path)
+    except OSError, e:
+        if e.errno == errno.EEXIST:
+            pass
+        else:
+            raise
+
+def maketemp():
+    tmp = os.path.join(os.path.dirname(__file__), 'tmp')
+    mkdir(tmp)
+    me = os.path.splitext(os.path.basename(__file__))[0]
+    tmp = os.path.join(tmp, me)
+    mkdir(tmp)
+    return tmp
+
+def writeFile(path, content):
+    tmp = '%s.tmp' % path
+    f = file(tmp, 'w')
+    try:
+        f.write(content)
+    finally:
+        f.close()
+    os.rename(tmp, path)
+
+def _key(s):
+    return ''.join(s.split('\n')).strip()
+
+KEY_1 = _key("""
+ssh-rsa +v5XLsUrLsHOKy7Stob1lHZM17YCCNXplcKfbpIztS2PujyixOaBev1ku6H6ny
+gUXfuYVzY+PmfTLviSwD3UETxEkR/jlBURACDQARJdUxpgt9XG2Lbs8bhOjonAPapxrH0o
+9O8R0Y6Pm1Vh+H2U0B4UBhPgEframpeJYedijBxBV5aq3yUvHkXpcjM/P0gsKqr036k= j
+unk@gunk
+""")
+
+KEY_2 = _key("""
+ssh-rsa 4BX2TxZoD3Og2zNjHwaMhVEa5/NLnPcw+Z02TDR0IGJrrqXk7YlfR3oz+Wb/Eb
+Ctli20SoWY0Ur8kBEF/xR4hRslZ2U8t0PAJhr8cq5mifhok/gAdckmSzjD67QJ68uZbga8
+ZwIAo7y/BU7cD3Y9UdVZykG34NiijHZLlCBo/TnobXjFIPXvFbfgQ3y8g+akwocFVcQ= f
+roop@snoop
+""")
+
+class ReadKeys_Test(object):
+    def test_empty(self):
+        tmp = maketemp()
+        empty = os.path.join(tmp, 'empty')
+        mkdir(empty)
+        gen = ssh.readKeys(keydir=empty)
+        assert_raises(StopIteration, gen.next)
+
+    def test_ignore_dot(self):
+        tmp = maketemp()
+        keydir = os.path.join(tmp, 'ignore_dot')
+        mkdir(keydir)
+        writeFile(os.path.join(keydir, '.jdoe.pub'), KEY_1+'\n')
+        gen = ssh.readKeys(keydir=keydir)
+        assert_raises(StopIteration, gen.next)
+
+    def test_ignore_nonpub(self):
+        tmp = maketemp()
+        keydir = os.path.join(tmp, 'ignore_dot')
+        mkdir(keydir)
+        writeFile(os.path.join(keydir, 'jdoe.xub'), KEY_1+'\n')
+        gen = ssh.readKeys(keydir=keydir)
+        assert_raises(StopIteration, gen.next)
+
+    def test_one(self):
+        tmp = maketemp()
+        keydir = os.path.join(tmp, 'one')
+        mkdir(keydir)
+        writeFile(os.path.join(keydir, 'jdoe.pub'), KEY_1+'\n')
+
+        gen = ssh.readKeys(keydir=keydir)
+        eq(gen.next(), ('jdoe', KEY_1))
+        assert_raises(StopIteration, gen.next)
+
+    def test_two(self):
+        tmp = maketemp()
+        keydir = os.path.join(tmp, 'two')
+        mkdir(keydir)
+        writeFile(os.path.join(keydir, 'jdoe.pub'), KEY_1+'\n')
+        writeFile(os.path.join(keydir, 'wsmith.pub'), KEY_2+'\n')
+
+        gen = ssh.readKeys(keydir=keydir)
+        got = frozenset(gen)
+
+        eq(got,
+           frozenset([
+            ('jdoe', KEY_1),
+            ('wsmith', KEY_2),
+            ]))
+
+class GenerateAuthorizedKeys_Test(object):
+    def test_simple(self):
+        def k():
+            yield ('jdoe', KEY_1)
+            yield ('wsmith', KEY_2)
+        gen = ssh.generateAuthorizedKeys(k())
+        eq(gen.next(), ssh.COMMENT)
+        eq(gen.next(), (
+            'command="gitosis-serve jdoe",no-port-forwarding,no-X11-f'
+            +'orwarding,no-agent-forwarding,no-pty %s' % KEY_1))
+        eq(gen.next(), (
+            'command="gitosis-serve wsmith",no-port-forwarding,no-X11'
+            +'-forwarding,no-agent-forwarding,no-pty %s' % KEY_2))
+        assert_raises(StopIteration, gen.next)
+
+
+class FilterAuthorizedKeys_Test(object):
+    def run(self, s):
+        f = StringIO(s)
+        lines = ssh.filterAuthorizedKeys(f)
+        got = ''.join(['%s\n' % line for line in lines])
+        return got
+
+    def check_no_change(self, s):
+        got = self.run(s)
+        eq(got, s)
+
+    def test_notFiltered_comment(self):
+        self.check_no_change('#comment\n')
+
+    def test_notFiltered_junk(self):
+        self.check_no_change('junk\n')
+
+    def test_notFiltered_key(self):
+        self.check_no_change('%s\n' % KEY_1)
+
+    def test_notFiltered_keyWithCommand(self):
+        s = '''\
+command="faketosis-serve wsmith",no-port-forwarding,no-X11-forwardin\
+g,no-agent-forwarding,no-pty %(key_1)s
+''' % dict(key_1=KEY_1)
+        self.check_no_change(s)
+
+
+    def test_filter_autogeneratedComment_backwardsCompat(self):
+        got = self.run('### autogenerated by gitosis, DO NOT EDIT\n')
+        eq(got, '')
+
+    def test_filter_autogeneratedComment_current(self):
+        got = self.run(ssh.COMMENT+'\n')
+        eq(got, '')
+
+    def test_filter_simple(self):
+        s = '''\
+command="gitosis-serve wsmith",no-port-forwarding,no-X11-forwardin\
+g,no-agent-forwarding,no-pty %(key_1)s
+''' % dict(key_1=KEY_1)
+        got = self.run(s)
+        eq(got, '')
+
+    def test_filter_withPath(self):
+        s = '''\
+command="/foo/bar/baz/gitosis-serve wsmith",no-port-forwarding,no-X11-forwardin\
+g,no-agent-forwarding,no-pty %(key_1)s
+''' % dict(key_1=KEY_1)
+        got = self.run(s)
+        eq(got, '')
+
+
+class WriteAuthorizedKeys_Test(object):
+    def test_simple(self):
+        tmp = maketemp()
+        path = os.path.join(tmp, 'authorized_keys')
+        oldfp = StringIO('''\
+# foo
+bar
+### autogenerated by gitosis, DO NOT EDIT
+command="/foo/bar/baz/gitosis-serve wsmith",no-port-forwarding,\
+no-X11-forwarding,no-agent-forwarding,no-pty %(key_2)s
+baz
+''' % dict(key_2=KEY_2))
+        keydir = os.path.join(tmp, 'one')
+        mkdir(keydir)
+        writeFile(os.path.join(keydir, 'jdoe.pub'), KEY_1+'\n')
+
+        ssh.writeAuthorizedKeys(
+            oldfp=oldfp, newpath=path, keydir=keydir)
+
+        f = file(path)
+        try:
+            got = f.read()
+        finally:
+            f.close()
+
+        eq(got, '''\
+# foo
+bar
+baz
+### autogenerated by gitosis, DO NOT EDIT
+command="gitosis-serve jdoe",no-port-forwarding,\
+no-X11-forwarding,no-agent-forwarding,no-pty %(key_1)s
+''' % dict(key_1=KEY_1))

setup.py

@@ -0,0 +1,21 @@
+#!/usr/bin/python
+from setuptools import setup, find_packages
+setup(
+    name = "gitosis",
+    version = "0.1",
+    packages = find_packages(),
+
+    author = "Tommi Virtanen",
+    author_email = "tv@eagain.net",
+    description = "software for hosting git repositories",
+    license = "GPL",
+    keywords = "git scm version-control ssh",
+    url = "http://eagain.net/software/gitosis/",
+
+    entry_points = {
+        'console_scripts': [
+            'gitosis-ssh = gitosis.ssh:main',
+            'gitosis-serve = gitosis.serve:main',
+            ],
+        },
+    )