/**
* Git P2P Protocol Handler for Pear-Git
*
* Handles git smart protocol over Hyperswarm connections:
* - git-upload-pack (clone/fetch)
* - git-receive-pack (push)
* - pgh-admin (P2P admin commands)
*
* Protocol:
* 1. Client connects to topic (derived from server key + repo name)
* 2. Client sends: service\0repoName\0 (e.g., "git-upload-pack\0myrepo\0")
* 3. Server verifies access, spawns git process
* 4. Bidirectional proxy: client <-> git process
*
* Admin Protocol (pgh-admin):
* 1. Client connects to __admin__ topic
* 2. Client sends: pgh-admin\0admin\0
* 3. Client sends: JSON request + newline
* 4. Server authenticates via NoiseKeyProvider
* 5. Server sends: JSON response + newline
* 6. Connection closes
*/
import { spawn } from 'bare-subprocess'
import path from 'bare-path'
import fs from 'bare-fs'
import { NoiseKeyProvider } from '../../auth/providers/noise-key.js'
import { autoDeployAfterPush } from '../../ci/deploy-hook.js'
import { refreshRepoAfterPush } from '../../routes/git.js'
export class GitP2PHandler {
constructor (options = {}) {
this.reposPath = options.reposPath
this.authManager = options.authManager || null
this.repoRegistry = options.repoRegistry || null
this.registry = options.registry || null // Hyperbee for peer-access lookups
this.topicManager = options.topicManager || null
this.serverIdentity = options.serverIdentity || null
this.backendManager = options.backendManager || null
this.activeConnections = new Map()
// Initialize Noise key provider for P2P auth
if (this.authManager) {
this.noiseKeyProvider = new NoiseKeyProvider(this.authManager.identityRegistry)
}
}
/**
* Handle incoming P2P connection
* @param {Stream} conn - Hyperswarm connection
* @param {Object} peerInfo - Peer information
*/
async handleConnection (conn, peerInfo) {
const connId = peerInfo.publicKey.toString('hex').slice(0, 8)
console.log(`[GitP2P] Connection from ${connId}`)
// No connection-level timeout — git transfers can run for minutes on large
// repos. Backpressure keeps data flowing; connection drops are detected by
// the stream's built-in error/close events.
let headerReceived = false
let buffer = Buffer.alloc(0)
// Header parser - will be removed after header is received
const headerHandler = async (chunk) => {
if (headerReceived) return // Safety check
buffer = Buffer.concat([buffer, chunk])
// Look for double null terminator (service\0repo\0)
const firstNull = buffer.indexOf(0)
if (firstNull === -1) return
const secondNull = buffer.indexOf(0, firstNull + 1)
if (secondNull === -1) return
headerReceived = true
// CRITICAL: Remove this listener before setting up pipes
conn.off('data', headerHandler)
const service = buffer.slice(0, firstNull).toString('utf-8')
const repoName = buffer.slice(firstNull + 1, secondNull).toString('utf-8')
const remaining = buffer.slice(secondNull + 1)
console.log(`[GitP2P] ${connId}: ${service} ${repoName}`)
try {
// Route to appropriate handler
if (service === 'pgh-admin') {
await this._handleAdminCommand(conn, peerInfo, remaining)
} else if (service === 'pgh-browse') {
await this._handleBrowseRequest(conn, peerInfo, repoName, remaining)
// After browse completes, re-enter header parsing for connection reuse.
// BrowseClient keeps connections open and sends multiple requests.
if (!conn.destroyed) {
headerReceived = false
buffer = Buffer.alloc(0)
conn.on('data', headerHandler)
}
} else {
await this._handleGitService(conn, service, repoName, remaining, peerInfo)
}
} catch (err) {
console.error(`[GitP2P] Error: ${err.message}`)
try { this._sendError(conn, err.message) } catch {}
if (!conn.destroyed) conn.destroy()
}
}
conn.on('data', headerHandler)
conn.on('error', (err) => {
console.error(`[GitP2P] Connection error ${connId}: ${err.message}`)
})
conn.on('close', () => {
console.log(`[GitP2P] Connection closed ${connId}`)
this._cleanupConnection(connId)
})
this.activeConnections.set(connId, { conn, peerInfo })
}
// ==================== Admin Command Handler ====================
/**
* Handle P2P admin commands
* Protocol: newline-delimited JSON request → JSON response
*/
async _handleAdminCommand (conn, peerInfo, initialData) {
const peerKeyHex = peerInfo.publicKey.toString('hex')
const connId = peerKeyHex.slice(0, 8)
// Authenticate via NoiseKeyProvider
if (!this.noiseKeyProvider) {
this._sendAdminResponse(conn, { error: 'Admin auth not configured' })
return
}
const auth = await this.noiseKeyProvider.authenticateByKey(peerKeyHex)
// Also check peer-access for admin (fallback for peers registered via invites)
let isAdmin = false
if (auth) {
const perms = this.authManager.formatUserPermissions(auth.user, null)
isAdmin = perms.admin
}
if (!auth || !isAdmin) {
console.log(`[GitP2P] Admin denied for ${connId}`)
this._sendAdminResponse(conn, { error: 'Admin access denied' })
return
}
console.log(`[GitP2P] Admin auth OK for ${connId} (user: ${auth.userId})`)
// Accumulate request data until newline
let requestBuffer = initialData || Buffer.alloc(0)
const MAX_REQUEST_SIZE = 64 * 1024 // 64KB max
// Check if initial data already has a complete request
const newlineIdx = requestBuffer.indexOf(10) // \n
if (newlineIdx !== -1) {
await this._dispatchAdminCommand(conn, auth, requestBuffer.slice(0, newlineIdx).toString('utf-8'))
return
}
// Wait for more data
let requestTimeout = null
const dataHandler = async (chunk) => {
requestBuffer = Buffer.concat([requestBuffer, chunk])
if (requestBuffer.length > MAX_REQUEST_SIZE) {
conn.off('data', dataHandler)
clearTimeout(requestTimeout)
this._sendAdminResponse(conn, { error: 'Request too large' })
return
}
const nl = requestBuffer.indexOf(10)
if (nl !== -1) {
conn.off('data', dataHandler)
clearTimeout(requestTimeout)
await this._dispatchAdminCommand(conn, auth, requestBuffer.slice(0, nl).toString('utf-8'))
}
}
conn.on('data', dataHandler)
if (typeof conn.resume === 'function') conn.resume()
// Timeout after 30 seconds
requestTimeout = setTimeout(() => {
conn.off('data', dataHandler)
if (!conn.destroyed) {
this._sendAdminResponse(conn, { error: 'Request timeout' })
}
}, 30000)
conn.once('close', () => clearTimeout(requestTimeout))
conn.once('error', () => clearTimeout(requestTimeout))
}
/**
* Parse and dispatch admin command
*/
async _dispatchAdminCommand (conn, auth, requestStr) {
let request
try {
request = JSON.parse(requestStr)
} catch {
this._sendAdminResponse(conn, { error: 'Invalid JSON' })
return
}
const { command } = request
console.log(`[GitP2P] Admin command: ${command} from ${auth.userId}`)
try {
let result
switch (command) {
case 'create-repo':
result = await this._adminCreateRepo(request, auth)
break
case 'list-repos':
result = await this._adminListRepos(request, auth)
break
case 'grant-access':
result = await this._adminGrantAccess(request, auth)
break
case 'revoke-access':
result = await this._adminRevokeAccess(request, auth)
break
case 'server-info':
result = await this._adminServerInfo(request, auth)
break
case 'whoami':
result = await this._adminWhoami(request, auth)
break
case 'sync-repo':
result = await this._adminSyncRepo(request, auth)
break
default:
result = { error: `Unknown command: ${command}` }
}
this._sendAdminResponse(conn, result)
} catch (err) {
console.error(`[GitP2P] Admin command error:`, err.message)
this._sendAdminResponse(conn, { error: err.message })
}
}
/**
* Send JSON response and close connection
*/
_sendAdminResponse (conn, data) {
if (conn.destroyed) return
conn.write(JSON.stringify(data) + '\n')
conn.end()
}
// ==================== Admin Command Implementations ====================
/**
* create-repo: Create a new bare git repository
*/
async _adminCreateRepo (request, auth) {
const { name, fromPath, visibility, category } = request
// Validate name
if (!name || !/^[a-zA-Z0-9._-]+$/.test(name)) {
return { error: 'Invalid repo name. Use alphanumeric, dots, hyphens, underscores.' }
}
if (name.startsWith('__') || name.startsWith('.')) {
return { error: `Reserved name: ${name}` }
}
// Check if exists
const existing = await this.repoRegistry.getRepo(name)
if (existing) {
return { error: `Repository already exists: ${name}` }
}
// Determine repo path
const repoPath = path.join(this.reposPath, name + '.git')
if (fromPath) {
// Clone from existing path
try {
await fs.promises.stat(fromPath)
} catch {
return { error: `Source path not found: ${fromPath}` }
}
await this.repoRegistry.execGit(this.reposPath, [
'clone', '--bare', fromPath, repoPath
])
console.log(`[GitP2P] Cloned ${fromPath} -> ${repoPath}`)
} else {
// Create empty bare repo
await fs.promises.mkdir(repoPath, { recursive: true })
await this.repoRegistry.execGit(repoPath, ['init', '--bare'])
console.log(`[GitP2P] Created empty repo: ${repoPath}`)
}
// Get repo info and register
const repoInfo = await this.repoRegistry.getRepoInfo(repoPath, name)
if (!repoInfo) {
return { error: 'Failed to get repo info after creation' }
}
repoInfo.visibility = visibility || 'public'
repoInfo.owner = auth.userId
if (category) repoInfo.category = category
await this.repoRegistry.register(repoInfo)
// Update in-memory cache so HTTP clients see it immediately
if (this.server && this.server.repos) {
this.server.repos.push(repoInfo)
}
// Announce on DHT
if (this.topicManager && repoInfo.visibility !== 'private') {
await this.topicManager.announceRepo(name)
}
// Build punch URL
let punchUrl = null
if (this.topicManager) {
punchUrl = this.topicManager.getPunchUrl(name)
}
return {
success: true,
name,
path: repoPath,
punchUrl,
visibility: repoInfo.visibility,
commits: repoInfo.commits || 0,
branches: repoInfo.branches || 0
}
}
/**
* list-repos: List all repositories
*/
async _adminListRepos (request, auth) {
const repos = await this.repoRegistry.listAll()
return {
repos: repos.map(r => ({
name: r.name,
visibility: r.visibility || 'public',
commits: r.commits || 0,
branches: r.branches || 0,
category: r.category || null,
owner: r.owner || null,
punchUrl: this.topicManager ? this.topicManager.getPunchUrl(r.name) : null
})),
count: repos.length
}
}
/**
* grant-access: Link a peer's Noise key and grant repo permissions
*/
async _adminGrantAccess (request, auth) {
const { repoName, peerKey, permissions } = request
if (!repoName || !peerKey) {
return { error: 'repoName and peerKey are required' }
}
if (!/^[0-9a-f]{64}$/i.test(peerKey)) {
return { error: 'peerKey must be a 64-char hex string' }
}
// Verify repo exists
const repo = await this.repoRegistry.getRepo(repoName)
if (!repo) {
return { error: `Repository not found: ${repoName}` }
}
const perms = Array.isArray(permissions) ? permissions : ['read', 'write']
// Store peer-access record (keyed by Noise key)
const accessKey = `peer-access:${repoName}:${peerKey}`
await this.registry.put(accessKey, {
repoName,
peerKey,
permissions: perms,
grantedBy: auth.userId,
grantedAt: Date.now()
})
console.log(`[GitP2P] Granted ${perms.join(',')} on ${repoName} to ${peerKey.slice(0, 8)}...`)
return { success: true, repoName, peerKey: peerKey.slice(0, 16) + '...', permissions: perms }
}
/**
* revoke-access: Remove a peer's access to a repo
*/
async _adminRevokeAccess (request, auth) {
const { repoName, peerKey } = request
if (!repoName || !peerKey) {
return { error: 'repoName and peerKey are required' }
}
const accessKey = `peer-access:${repoName}:${peerKey}`
const entry = await this.registry.get(accessKey)
if (!entry) {
return { error: 'No access record found' }
}
await this.registry.del(accessKey)
console.log(`[GitP2P] Revoked access on ${repoName} for ${peerKey.slice(0, 8)}...`)
return { success: true, repoName, peerKey: peerKey.slice(0, 16) + '...' }
}
/**
* server-info: Return server status
*/
async _adminServerInfo (request, auth) {
const repos = await this.repoRegistry.listAll()
const serverKey = this.serverIdentity ? this.serverIdentity.getPublicKeyHex() : null
return {
serverKey,
repoCount: repos.length,
activeConnections: this.activeConnections.size,
topics: this.topicManager ? this.topicManager.listTopics() : []
}
}
/**
* whoami: Return peer identity and permissions
*/
async _adminWhoami (request, auth) {
return {
userId: auth.userId,
identityId: auth.identityId,
method: auth.method,
permissions: this.authManager.formatUserPermissions(auth.user, null),
peerKey: auth.user?.identities?.find(id => id.startsWith('noise:'))?.slice(6) || null
}
}
/**
* sync-repo: Fetch a repo from a remote peer (push-triggered sync)
*/
async _adminSyncRepo (request, auth) {
const { repoName, serverKey } = request
if (!repoName) return { error: 'repoName required' }
const repoPath = path.join(this.reposPath, `${repoName}.git`)
try {
await fs.promises.stat(repoPath)
} catch {
return { error: `Repository ${repoName} not found locally` }
}
if (!serverKey) return { error: 'serverKey required' }
const fetchUrl = `punch://${serverKey}/${repoName}`
console.log(`[GitP2P] Sync-repo: fetching ${repoName} from ${serverKey.slice(0, 8)}`)
try {
await new Promise((resolve, reject) => {
let done = false
// Fast-forward only — never rewind refs from a stale peer.
// Without '+' prefix, git refuses non-fast-forward updates,
// preventing a stale federation peer from overwriting newer commits.
const child = spawn('git', [
'-C', repoPath, 'fetch', fetchUrl,
'refs/*:refs/*', '--no-tags'
])
const timer = setTimeout(() => {
if (done) return
done = true
try { child.kill('SIGKILL') } catch {}
reject(new Error('fetch timeout'))
}, 120000)
let stderr = ''
child.stderr.on('data', (d) => { stderr += d.toString() })
child.on('exit', (code) => {
if (done) return
done = true
clearTimeout(timer)
if (code === 0) resolve()
else reject(new Error(`exit ${code}: ${stderr.trim()}`))
})
child.on('error', (err) => {
if (done) return
done = true
clearTimeout(timer)
reject(err)
})
})
console.log(`[GitP2P] Sync-repo: ${repoName} fetched OK`)
// Log sync event
if (this.registry) {
try {
await this.registry.put(`sync-log:${Date.now()}`, {
repoName, serverKey, direction: 'received', success: true, timestamp: Date.now()
})
} catch {}
}
// Trigger self-update if this repo is pear-git itself
if (this.server && this.server.gitSyncClientManager) {
const cm = this.server.gitSyncClientManager
if (cm.appRepoName && repoName === cm.appRepoName) {
cm._selfUpdate(repoPath).catch(err => {
console.error(`[GitP2P] Self-update failed: ${err.message}`)
})
}
}
return { ok: true, repoName }
} catch (err) {
console.error(`[GitP2P] Sync-repo: ${repoName} failed: ${err.message}`)
return { error: err.message }
}
}
// ==================== Browse Handler (read-only, no auth) ====================
/**
* Handle P2P browse requests — read-only repo browsing over federation
* No auth required (federation = full read access)
*
* Protocol: newline-delimited JSON request → JSON response, then close
*/
async _handleBrowseRequest (conn, peerInfo, repoName, initialData) {
const connId = peerInfo.publicKey.toString('hex').slice(0, 8)
// CRITICAL: Install data listener SYNCHRONOUSLY before any await.
// Between headerHandler removal and this function, there's no listener.
// The JSON request may arrive during any await — must not lose data.
var requestBuffer = initialData || Buffer.alloc(0)
var MAX_REQUEST_SIZE = 64 * 1024
var gotRequest = null
var requestPromise = new Promise(function (resolve) { gotRequest = resolve })
var requestTimeout = setTimeout(function () {
conn.off('data', dataHandler)
gotRequest(null)
}, 30000)
function dataHandler (chunk) {
requestBuffer = Buffer.concat([requestBuffer, chunk])
if (requestBuffer.length > MAX_REQUEST_SIZE) {
conn.off('data', dataHandler)
clearTimeout(requestTimeout)
gotRequest(null)
return
}
var nl = requestBuffer.indexOf(10)
if (nl !== -1) {
conn.off('data', dataHandler)
clearTimeout(requestTimeout)
gotRequest(requestBuffer.slice(0, nl).toString('utf-8'))
}
}
// Check if initialData already has the request
var newlineIdx = requestBuffer.indexOf(10)
if (newlineIdx !== -1) {
clearTimeout(requestTimeout)
gotRequest(requestBuffer.slice(0, newlineIdx).toString('utf-8'))
} else {
conn.on('data', dataHandler)
if (typeof conn.resume === 'function') conn.resume()
}
conn.once('close', function () { clearTimeout(requestTimeout) })
conn.once('error', function () { clearTimeout(requestTimeout) })
// Now safe to do async validation while data accumulates
// Skip backendManager/repo checks for __admin__ pseudo-repo (server-level commands)
if (!this.backendManager && repoName !== '__admin__') {
clearTimeout(requestTimeout)
conn.off('data', dataHandler)
this._sendBrowseResponse(conn, { error: 'Browse not available' })
return
}
if (this.repoRegistry && repoName !== '__admin__') {
var repo = await this.repoRegistry.getRepo(repoName)
if (!repo) {
clearTimeout(requestTimeout)
conn.off('data', dataHandler)
this._sendBrowseResponse(conn, { error: 'Repository not found: ' + repoName })
return
}
}
// Wait for the JSON request to arrive
var requestStr = await requestPromise
if (!requestStr) {
if (!conn.destroyed) this._sendBrowseResponse(conn, { error: 'Request timeout' })
return
}
await this._dispatchBrowseCommand(conn, repoName, requestStr, connId)
}
/**
* Dispatch a browse command
*/
async _dispatchBrowseCommand (conn, repoName, requestStr, connId) {
let request
try {
request = JSON.parse(requestStr)
} catch {
this._sendBrowseResponse(conn, { error: 'Invalid JSON' })
return
}
const { command, branch, path: filePath, count } = request
console.log(`[GitP2P] Browse: ${command} ${repoName}${filePath ? '/' + filePath : ''} from ${connId}`)
try {
let result
switch (command) {
case 'listFiles':
result = await this.backendManager.listFiles(repoName, branch || 'HEAD', filePath || '')
break
case 'readFile':
result = { content: await this.backendManager.readFile(repoName, filePath || '') }
break
case 'getCommits':
result = await this.backendManager.getCommits(repoName, branch || 'HEAD', count || 20)
break
case 'getReadme':
result = await this.backendManager.getReadme(repoName, branch || 'HEAD')
break
case 'getBranches':
result = await this.backendManager.getBranches(repoName)
break
case 'getStats':
result = await this.backendManager.getStats(repoName)
break
case 'repoPage': {
// Fetch everything needed for the repo viewer in one round-trip
const b = branch || 'HEAD'
const [rpFiles, rpCommits, rpReadme, rpStats, rpBranches] = await Promise.all([
this.backendManager.listFiles(repoName, b, filePath || ''),
this.backendManager.getCommits(repoName, b, count || 20),
this.backendManager.getReadme(repoName, b),
this.backendManager.getStats(repoName),
this.backendManager.getBranches(repoName)
])
result = { files: rpFiles, commits: rpCommits, readme: rpReadme, stats: rpStats, branches: rpBranches }
break
}
case 'serverCatalog': {
// Return this server's groups and repos for remote catalog browsing
const catalogGroups = []
const nsBackend = this.server && (this.server.namespaceRegistry || this.server.namespaceManager)
if (nsBackend && nsBackend.list) {
const allNs = await nsBackend.list({ type: 'group' })
for (const ns of allNs) {
catalogGroups.push({
name: ns.name,
displayName: ns.displayName || ns.name,
description: ns.description || '',
repos: ns.repos || [],
memberCount: (ns.members || []).length,
source: ns.source || 'local'
})
}
}
const catalogRepos = (this.server && this.server.repos || []).map(r => ({
name: r.name,
commits: r.commits || 0,
branches: r.branches || 0,
visibility: r.visibility || 'public',
category: r.category || null
}))
const serverKey = this.serverIdentity ? this.serverIdentity.getPublicKeyHex() : null
const serverName = (this.server && this.server.defaultNamespace) || 'unknown'
result = { serverName, serverKey, groups: catalogGroups, repos: catalogRepos }
break
}
case 'syncRepo': {
// Federation push-triggered sync: fetch a repo from the requesting server
// Only allowed from known federation peers
const { repoName: syncRepoName, serverKey: senderKey } = request
if (!syncRepoName || !senderKey) {
result = { error: 'syncRepo requires repoName and serverKey' }
break
}
// Verify sender is a known federation peer
let isFedPeer = false
if (this.registry) {
try {
const fedStream = this.registry.createReadStream({ gt: 'federation:', lt: 'federation:\xff' })
for await (const entry of fedStream) {
const data = typeof entry.value === 'string' ? JSON.parse(entry.value) : entry.value
if (data.status === 'active' && data.serverKey === senderKey) {
isFedPeer = true
break
}
}
} catch {}
}
if (!isFedPeer) {
result = { error: 'Not a known federation peer' }
break
}
result = await this._adminSyncRepo({ repoName: syncRepoName, serverKey: senderKey }, null)
break
}
default:
result = { error: `Unknown browse command: ${command}` }
}
if (result && result.error) {
this._sendBrowseResponse(conn, { ok: false, error: result.error })
} else {
this._sendBrowseResponse(conn, { ok: true, data: result })
}
} catch (err) {
console.error(`[GitP2P] Browse error:`, err.message)
this._sendBrowseResponse(conn, { ok: false, error: err.message })
}
}
_sendBrowseResponse (conn, data) {
if (conn.destroyed) return
conn.write(JSON.stringify(data) + '\n')
// NOTE: Do NOT call conn.end() here.
// The connection is reused for multiple browse requests (see headerHandler
// re-attach at line 89-95). Calling end() closes the write side, which
// silently drops all subsequent responses and causes 30s client timeouts.
}
// ==================== Git Service Handler ====================
/**
* Handle a git service request
*/
async _handleGitService (conn, service, repoName, initialData, peerInfo) {
// Validate service
const allowedServices = ['git-upload-pack', 'git-receive-pack']
if (!allowedServices.includes(service)) {
throw new Error(`Unknown service: ${service}`)
}
// Validate repo exists and is accessible
const repoPath = await this._validateRepo(repoName, service, peerInfo)
// Spawn git process
console.log(`[GitP2P] Spawning: ${service} ${repoPath}`)
const gitProcess = spawn(service, [repoPath], {
stdio: ['pipe', 'pipe', 'pipe']
})
// Send any remaining data from initial packet
if (initialData.length > 0) {
console.log(`[GitP2P] Sending initial data: ${initialData.length} bytes`)
gitProcess.stdin.write(initialData)
}
// Handle errors on all streams to prevent uncaught exceptions
gitProcess.stdin.on('error', (err) => {
if (err.code !== 'EPIPE' && err.code !== 'ECONNRESET') {
console.error(`[GitP2P] git stdin error: ${err.message}`)
}
})
gitProcess.stdout.on('error', (err) => {
if (err.code !== 'EPIPE' && err.code !== 'ECONNRESET') {
console.error(`[GitP2P] git stdout error: ${err.message}`)
}
})
// Manual bidirectional proxy instead of .pipe().
// Explicit resume() after adding data listener — removing the
// headerHandler may leave the noise stream paused in streamx.
//
// Backpressure: when write() returns false, pause the source.
// Resume on 'drain'. Without this, large repos (1+ GB) overflow
// the send buffer and cause "unexpected disconnect" / "bad pack header".
conn.on('data', (chunk) => {
console.log(`[GitP2P] conn->git: ${chunk.length} bytes`)
if (!gitProcess.stdin.write(chunk)) {
conn.pause()
}
})
gitProcess.stdin.on('drain', () => conn.resume())
let firstChunk = true
gitProcess.stdout.on('data', (chunk) => {
console.log(`[GitP2P] git->conn: ${chunk.length} bytes`)
if (firstChunk) {
// Log first 200 bytes of ref advertisement for debugging stale refs
const preview = chunk.slice(0, 200).toString('utf-8').replace(/\0/g, '\\0')
console.log(`[GitP2P] REFS: ${preview}`)
firstChunk = false
}
if (!conn.write(chunk)) {
gitProcess.stdout.pause()
}
})
conn.on('drain', () => gitProcess.stdout.resume())
// Ensure conn is in flowing mode after headerHandler was removed
if (typeof conn.resume === 'function') conn.resume()
conn.on('end', () => {
console.log('[GitP2P] conn ended, closing git stdin')
gitProcess.stdin.end()
})
gitProcess.stderr.on('data', (data) => {
console.error(`[GitP2P] git stderr: ${data.toString()}`)
})
gitProcess.on('exit', async (code) => {
console.log(`[GitP2P] git process exited with code ${code}`)
// After successful push, update Autobase if repo is linked
if (code === 0 && service === 'git-receive-pack' && this.server && this.server.linkedRepos) {
const linked = this.server.linkedRepos.get(repoName)
if (linked) {
try {
// Resolve userId from peer's Noise key
const peerKeyHex = peerInfo.publicKey.toString('hex')
let userId = peerKeyHex.slice(0, 16) // fallback
if (this.noiseKeyProvider) {
const auth = await this.noiseKeyProvider.authenticateByKey(peerKeyHex)
if (auth) userId = auth.userId
}
// Read current refs from git and sync to Autobase
const refOutput = await new Promise((resolve, reject) => {
const proc = spawn('git', ['-C', repoPath, 'for-each-ref', '--format=%(refname) %(objectname)', 'refs/heads/', 'refs/tags/'], {
stdio: ['ignore', 'pipe', 'pipe']
})
const chunks = []
proc.stdout.on('data', (c) => chunks.push(c))
proc.on('exit', (c) => c === 0 ? resolve(Buffer.concat(chunks).toString('utf-8')) : resolve(''))
proc.on('error', () => resolve(''))
})
for (const line of refOutput.trim().split('\n')) {
if (!line) continue
const [ref, hash] = line.split(' ')
if (ref && hash) {
await linked.updateRef(ref, hash, userId)
}
}
console.log(`[GitP2P] Updated linked repo refs for ${repoName}`)
} catch (err) {
console.error(`[GitP2P] Failed to update linked repo refs:`, err.message)
}
}
}
// Auto-deploy after successful push (if configured, non-blocking)
if (code === 0 && service === 'git-receive-pack' && this.server) {
autoDeployAfterPush(this.server, repoName)
refreshRepoAfterPush(this.server, repoName)
// Broadcast ref update to federation peers via Protomux
if (this.server.gitSyncManager) {
import('../../git/ref-reader.js').then(({ readRefs }) => {
return readRefs(repoPath)
}).then(refs => {
const serverKey = this.serverIdentity ? this.serverIdentity.getPublicKeyHex() : null
this.server.gitSyncManager.broadcastRefUpdate(repoName, refs, serverKey)
console.log('[GitSync] Broadcast ref-update for ' + repoName)
}).catch(err => {
console.error('[GitSync] Broadcast failed:', err.message)
})
}
}
if (!conn.destroyed) conn.end()
})
gitProcess.on('error', (err) => {
console.error(`[GitP2P] git process error: ${err.message}`)
if (!conn.destroyed) conn.destroy()
})
// When connection drops, kill the git process to prevent orphans
conn.on('close', () => {
if (gitProcess.exitCode === null) {
gitProcess.kill()
}
})
}
/**
* Validate repository access using P2P peer identity
*
* Auth model:
* - Public repos: anyone can clone/fetch (git-upload-pack)
* - Private repos: only peers with peer-access records can clone/fetch
* - Push (git-receive-pack): requires peer-access with 'write' permission
* OR admin Noise key identity
* - Peer identity: conn.remotePublicKey (Noise protocol, verified)
*/
async _validateRepo (repoName, service, peerInfo) {
// Check if repo exists in registry
if (this.repoRegistry) {
const repo = await this.repoRegistry.getRepo(repoName)
if (!repo) {
throw new Error(`Repository not found: ${repoName}`)
}
const peerKeyHex = peerInfo.publicKey.toString('hex')
// For push (receive-pack), check write access
if (service === 'git-receive-pack') {
// Check peer-access records first
const access = await this._getPeerAccess(repoName, peerKeyHex)
if (access && access.permissions && access.permissions.includes('write')) {
console.log(`[GitP2P] Push authorized (peer-access) for ${peerKeyHex.slice(0, 8)}... on ${repoName}`)
} else {
// Fallback: check if peer is admin via NoiseKeyProvider
let isAdmin = false
if (this.noiseKeyProvider) {
const auth = await this.noiseKeyProvider.authenticateByKey(peerKeyHex)
if (auth) {
const perms = this.authManager.formatUserPermissions(auth.user, null)
isAdmin = perms.admin
}
}
if (!isAdmin) {
throw new Error(`Push denied: peer ${peerKeyHex.slice(0, 8)}... lacks write permission for ${repoName}`)
}
console.log(`[GitP2P] Push authorized (admin) for ${peerKeyHex.slice(0, 8)}... on ${repoName}`)
}
}
// For clone/fetch (upload-pack), check visibility
if (repo.visibility === 'private') {
const access = await this._getPeerAccess(repoName, peerKeyHex)
if (!access) {
// Admins can access private repos too
let isAdmin = false
if (this.noiseKeyProvider) {
const auth = await this.noiseKeyProvider.authenticateByKey(peerKeyHex)
if (auth) {
const perms = this.authManager.formatUserPermissions(auth.user, null)
isAdmin = perms.admin
}
}
if (!isAdmin) {
throw new Error(`Access denied: ${repoName} is private`)
}
}
console.log(`[GitP2P] Private access granted for ${peerKeyHex.slice(0, 8)}... on ${repoName}`)
}
// Use path from registry if available
if (repo.path) {
return repo.path
}
}
// Fallback: Construct repo path (with path traversal protection)
const repoPath = path.resolve(this.reposPath, repoName)
const base = path.resolve(this.reposPath)
if (!repoPath.startsWith(base + path.sep) && repoPath !== base) {
throw new Error('Invalid repository name')
}
if (!repoPath.endsWith('.git')) {
return repoPath + '.git'
}
return repoPath
}
/**
* Look up peer access from the Hyperbee registry
* Checks peer-access:{repoName}:{peerKey}
*/
async _getPeerAccess (repoName, peerKeyHex) {
if (!this.registry) return null
try {
const entry = await this.registry.get(`peer-access:${repoName}:${peerKeyHex}`)
return entry ? entry.value : null
} catch {
return null
}
}
/**
* Send error message to client
*/
_sendError (conn, message) {
// Git protocol error format
const errorLine = `ERR ${message}\n`
const length = (4 + errorLine.length).toString(16).padStart(4, '0')
conn.write(length + errorLine)
}
/**
* Clean up connection state
*/
_cleanupConnection (connId) {
this.activeConnections.delete(connId)
}
/**
* Get active connection count
*/
getConnectionCount () {
return this.activeConnections.size
}
/**
* Destroy all connections
*/
async destroy () {
for (const [id, { conn }] of this.activeConnections) {
conn.destroy()
}
this.activeConnections.clear()
}
}
/**
* Client-side git P2P helper
* Used by git-remote-pear
*/
export class GitP2PClient {
constructor (options = {}) {
this.petnames = options.petnames
}
/**
* Initiate a git operation over P2P
* @param {Stream} conn - Hyperswarm connection
* @param {string} service - git-upload-pack or git-receive-pack
* @param {string} repoName - Repository name
*/
async initiate (conn, service, repoName) {
// Send service header: service\0repo\0
const header = Buffer.concat([
Buffer.from(service, 'utf-8'),
Buffer.from([0]),
Buffer.from(repoName, 'utf-8'),
Buffer.from([0])
])
conn.write(header)
// Return connection for bidirectional communication
return conn
}
}
export default GitP2PHandler