diff --git a/README.md b/README.md index db68f0b..7b6dfb8 100644 --- a/README.md +++ b/README.md @@ -1 +1,32 @@ Share file with people on the same network + +## Direct transfers + +Devices that see each other's shares (same public IP) are usually on the same +local network. When both sides are browsers, clicking **Download** first tries +a direct WebRTC DataChannel between them, signaled over the existing +WebSocket: + +- Only LAN host candidates are used: mDNS `.local` names (Chrome, Safari) or + private / link-local addresses (Firefox doesn't hide host candidates behind + mDNS). No STUN, no TURN, no public address ever: the direct path can only + succeed on the actual LAN. The server enforces this too and drops any other + candidate. The only thing a peer in your namespace can learn is your private + LAN address. +- A session can only be opened toward the owner of a share visible in your own + namespace, i.e. exactly what `GET /share/:uuid` already allows. +- The channel is DTLS-encrypted end to end; the server only relays the + handshake and never sees the content. +- If the channel is not open within 5 seconds (guest Wi-Fi client isolation, + VLANs, different networks...), the download silently falls back to the + server relay, exactly as before. `curl`, the QR code and `/all` always use + the relay. + +A service worker (`public/sw.js`) streams direct transfers straight into a +regular browser download, so file size is not limited by memory. Without it +(unsupported browser, insecure context), files up to 200 MB are saved from +memory and bigger ones use the relay. + +Server logs report how each direct session ended (`RTC session ... closed: +done|timeout|...`), which tells you how often the direct path actually works. + diff --git a/lib/dlcenter/client.rb b/lib/dlcenter/client.rb index ccfd6ee..6b512f7 100644 --- a/lib/dlcenter/client.rb +++ b/lib/dlcenter/client.rb @@ -1,4 +1,5 @@ require 'base64' +require 'ipaddr' module DLCenter # Input validation constants @@ -7,6 +8,13 @@ module DLCenter MAX_INLINE_CONTENT_LENGTH = 10_000 MAX_SHARES_PER_CLIENT = 100 MAX_CHUNK_SIZE = 2 * 1024 * 1024 # 2MB max chunk size + UUID_FORMAT = /\A[a-f0-9\-]{36}\z/i + + # WebRTC signaling limits + MAX_RTC_SESSIONS_PER_CLIENT = 8 + MAX_RTC_SDP_LENGTH = 16 * 1024 + MAX_RTC_CANDIDATE_LENGTH = 1024 + MAX_RTC_CANDIDATES_PER_SESSION = 64 class Client attr_reader :shares @@ -120,6 +128,7 @@ def initialize namespace, ws, &on_close @ws = ws @on_close = on_close @alive = true + @rtc_sessions = {} @heartbeat_timer = nil @timeout_timer = nil @@ -147,6 +156,7 @@ def initialize namespace, ws, &on_close guard("onclose") do puts "WS closed" stop_heartbeat + close_rtc_sessions @namespace.remove_client(self) @on_close.call if @on_close end @@ -209,6 +219,7 @@ def send_msg(msg, params={}) when :hello then send({type: :hello, text: params[:text]}.to_json) when :stream then send({type: :stream}.merge(params).to_json) when :stream_close then send({type: :stream_close, uuid: params[:uuid]}.to_json) + when :rtc then send(params.to_json) else raise "Invalid msg type #{msg} with params #{params}" end @@ -344,6 +355,127 @@ def handle_chunk(msg) return true end + # --- WebRTC signaling --------------------------------------------------- + # + # Two browsers on the same LAN can transfer directly over a DataChannel + # instead of relaying through us. The server only routes the handshake: + # a session can only be opened toward the owner of a share visible in the + # requester's own namespace (the same authorization as GET /share/:uuid), + # session ids are minted here, and peers never learn each other's client + # identity. Only LAN host candidates are relayed (mDNS names or private + # addresses, no STUN/TURN), so the direct path can only succeed on the + # actual local network and no public address is disclosed to anybody. + + attr_reader :rtc_sessions + + def valid_uuid?(uuid) + uuid.is_a?(String) && uuid.match?(UUID_FORMAT) + end + + def valid_rtc_sdp?(sdp) + sdp.is_a?(String) && !sdp.empty? && sdp.length <= MAX_RTC_SDP_LENGTH + end + + PRIVATE_RANGES = %w[10.0.0.0/8 172.16.0.0/12 192.168.0.0/16 169.254.0.0/16 fc00::/7 fe80::/10] + .map { |cidr| IPAddr.new(cidr) }.freeze + + # A host candidate that can only be reached from the local network: an + # mDNS name (Chrome, Safari) or a private / link-local address (Firefox + # doesn't obfuscate host candidates by default). Never a public address, + # never a server-reflexive or relay candidate. + def lan_address?(address) + return true if address.match?(/\.local\z/i) + ip = IPAddr.new(address) + PRIVATE_RANGES.any? { |range| range.include?(ip) } + rescue IPAddr::Error, ArgumentError + false + end + + # "candidate:
typ ..." + # An empty candidate string is the end-of-candidates marker. + def lan_candidate?(candidate) + return true if candidate.empty? + fields = candidate.split(' ') + fields[6] == 'typ' && fields[7] == 'host' && lan_address?(fields[4].to_s) + end + + # Drops any candidate line that is not a LAN host candidate, so that a + # modified client can't advertise a routable address through us. + def scrub_rtc_sdp(sdp) + sdp.each_line.reject do |line| + line.start_with?('a=candidate:') && !lan_candidate?(line.chomp.delete_prefix('a=')) + end.join + end + + def rtc_open_session(session, peer) + @rtc_sessions[session] = { peer: peer, candidates: 0 } + end + + def handle_rtc_offer(msg) + share_uuid = msg[:share] + return puts("Invalid RTC offer") unless valid_uuid?(share_uuid) && valid_rtc_sdp?(msg[:sdp]) + return puts("Too many RTC sessions") if @rtc_sessions.size >= MAX_RTC_SESSIONS_PER_CLIENT + share = @namespace.get_share_by_uuid(share_uuid) + owner = share&.client + # IOClient (curl uploads) has no browser to answer, and downloading + # one's own share directly makes no sense: both stay on the relay. + return puts("No RTC peer for share #{share_uuid}") unless owner.is_a?(WSClient) && owner != self + return puts("Peer has too many RTC sessions") if owner.rtc_sessions.size >= MAX_RTC_SESSIONS_PER_CLIENT + + session = SecureRandom.uuid + rtc_open_session(session, owner) + owner.rtc_open_session(session, self) + send_msg(:rtc, type: :rtc_session, session: session, share: share_uuid) + owner.send_msg(:rtc, type: :rtc_offer, session: session, share: share_uuid, sdp: scrub_rtc_sdp(msg[:sdp])) + end + + def handle_rtc_answer(msg) + state = @rtc_sessions[msg[:session]] + return unless state && valid_rtc_sdp?(msg[:sdp]) + state[:peer].send_msg(:rtc, type: :rtc_answer, session: msg[:session], sdp: scrub_rtc_sdp(msg[:sdp])) + end + + def handle_rtc_ice(msg) + state = @rtc_sessions[msg[:session]] + return unless state + candidate = msg[:candidate] + return puts("Invalid ICE candidate") unless candidate.is_a?(Hash) + candidate_str = candidate[:candidate] + return puts("Invalid ICE candidate") unless candidate_str.is_a?(String) && candidate_str.length <= MAX_RTC_CANDIDATE_LENGTH + return puts("Dropping non-LAN ICE candidate") unless lan_candidate?(candidate_str) + state[:candidates] += 1 + return puts("Too many ICE candidates") if state[:candidates] > MAX_RTC_CANDIDATES_PER_SESSION + + # Rebuild the candidate from whitelisted fields only. + forwarded = { candidate: candidate_str } + forwarded[:sdpMid] = candidate[:sdpMid][0, 64] if candidate[:sdpMid].is_a?(String) + forwarded[:sdpMLineIndex] = candidate[:sdpMLineIndex] if candidate[:sdpMLineIndex].is_a?(Integer) + forwarded[:usernameFragment] = candidate[:usernameFragment][0, 256] if candidate[:usernameFragment].is_a?(String) + state[:peer].send_msg(:rtc, type: :rtc_ice, session: msg[:session], candidate: forwarded) + end + + def handle_rtc_close(msg) + session = msg[:session] + state = @rtc_sessions.delete(session) + return unless state + # The reason ends up in the logs: keep it to a harmless token. + reason = msg[:reason].is_a?(String) ? msg[:reason].gsub(/[^\w\-]/, '')[0, 32] : 'unknown' + # Instrumentation: how often does the direct path actually work? + puts "RTC session #{session} closed: #{reason}" + state[:peer].rtc_peer_closed(session) + end + + def rtc_peer_closed(session) + return unless @rtc_sessions.delete(session) + send_msg(:rtc, type: :rtc_close, session: session) + end + + def close_rtc_sessions + sessions = @rtc_sessions + @rtc_sessions = {} + sessions.each { |session, state| state[:peer].rtc_peer_closed(session) } + end + def handle_ws_msg(msg) #puts msg case msg[:type] @@ -352,6 +484,10 @@ def handle_ws_msg(msg) when 'chunk' then handle_chunk(msg) when 'ping' then send({type: :pong}.to_json) when 'pong' then nil # liveness already recorded in onmessage + when 'rtc_offer' then handle_rtc_offer(msg) + when 'rtc_answer' then handle_rtc_answer(msg) + when 'rtc_ice' then handle_rtc_ice(msg) + when 'rtc_close' then handle_rtc_close(msg) else puts "Unkown msg : #{msg}" end end diff --git a/public/css/main.css b/public/css/main.css index dc6389e..99251ec 100644 --- a/public/css/main.css +++ b/public/css/main.css @@ -177,6 +177,13 @@ a:visited:hover { } } +.share__download--busy { + opacity: 0.8; + cursor: progress; + min-width: 9rem; + justify-content: center; +} + .share__remove { color: var(--red); cursor: pointer; diff --git a/public/index.html b/public/index.html index afca362..e2507f1 100644 --- a/public/index.html +++ b/public/index.html @@ -30,15 +30,16 @@ - + +
- - + + diff --git a/public/js/app.js b/public/js/app.js index 8b3c78c..ddf3175 100644 --- a/public/js/app.js +++ b/public/js/app.js @@ -6,6 +6,8 @@ function App() { const [localShares, setLocalShares] = useState({}); const [qrModal, setQrModal] = useState(null); const [textValue, setTextValue] = useState(''); + // Per-share download state: { status: connecting|direct|relay|done, progress } + const [downloads, setDownloads] = useState({}); const wsRef = useRef(null); const pingRef = useRef(null); const reconnectTimerRef = useRef(null); @@ -93,6 +95,42 @@ function App() { } }, []); + const setDownload = useCallback((uuid, value) => { + setDownloads(prev => { + const next = { ...prev }; + if (value) next[uuid] = value; else delete next[uuid]; + return next; + }); + }, []); + + // Download a remote share: try the direct browser-to-browser path first, + // fall back to the server relay (the plain /share/:uuid link) otherwise. + const downloadShare = useCallback((share) => { + const url = `${downloadHost}/share/${share.uuid}`; + const clearLater = () => setTimeout(() => setDownload(share.uuid, null), 2000); + const relay = () => { + setDownload(share.uuid, { status: 'relay' }); + relayDownload(url); + clearLater(); + }; + const ws = wsRef.current; + if (!rtcSupported() || !ws || ws.readyState !== WebSocket.OPEN) { + relay(); + return; + } + setDownload(share.uuid, { status: 'connecting' }); + rtcDownload(share, ws, { + onConnected: () => setDownload(share.uuid, { status: 'direct', progress: 0 }), + onProgress: (received, size) => setDownload(share.uuid, { status: 'direct', progress: size ? received / size : 1 }), + onDone: () => { setDownload(share.uuid, { status: 'done' }); clearLater(); }, + onAbort: () => setDownload(share.uuid, null), + onFallback: (reason) => { + console.log("direct download unavailable (" + reason + "), using relay"); + relay(); + } + }); + }, [downloadHost, setDownload]); + const handleStream = useCallback((msg) => { console.log("Should stream file " + msg.share + " to stream " + msg.uuid); // Use ref for synchronous lookup @@ -176,6 +214,7 @@ function App() { setConnected(false); setRemoteShares([]); streamAbortAll(); + rtcSocketClosed(); if (pingRef.current) { clearInterval(pingRef.current); pingRef.current = null; @@ -211,6 +250,13 @@ function App() { break; case "pong": break; + case "rtc_session": + case "rtc_offer": + case "rtc_answer": + case "rtc_ice": + case "rtc_close": + rtcHandleMessage(msg, ws, localSharesRef.current); + break; default: console.warn("Unknown message: " + msg.type); } @@ -218,6 +264,16 @@ function App() { }, [handleStream, registerLocalShares, scheduleReconnect]); setupRef.current = setupWebSocket; + // The service worker streams direct transfers straight to disk (see sw.js). + // Optional: without it, small files are saved from memory and big ones + // take the relay. + useEffect(() => { + if (!('serviceWorker' in navigator)) return; + navigator.serviceWorker.register('/sw.js').catch((err) => { + console.log("service worker unavailable: " + err.message); + }); + }, []); + useEffect(() => { setupWebSocket(); return () => { @@ -308,8 +364,10 @@ function App() { remoteShares={remoteShares} localShares={localShares} downloadHost={downloadHost} + downloads={downloads} onRemove={removeShare} onShowQR={setQrModal} + onDownload={downloadShare} /> {qrModal && ( diff --git a/public/js/components.js b/public/js/components.js index 5843d3f..427c5f9 100644 --- a/public/js/components.js +++ b/public/js/components.js @@ -53,10 +53,31 @@ function QRCodeModal({ share, downloadHost, onClose }) { ); } -function Share({ share, localShares, downloadHost, onRemove, onShowQR }) { +function downloadLabel(download) { + if (!download) return 'Download'; + switch (download.status) { + case 'connecting': return 'Connecting\u2026'; + case 'direct': return 'Direct ' + Math.round((download.progress || 0) * 100) + '%'; + case 'relay': return 'Downloading\u2026'; + case 'done': return 'Done'; + default: return 'Download'; + } +} + +function Share({ share, localShares, downloadHost, downloads, onRemove, onShowQR, onDownload }) { const canDelete = localShares[share.uuid]; const isLink = share.link; const displayName = share.name.length > 45 ? share.name.substring(0, 45) + '...' : share.name; + const download = downloads && downloads[share.uuid]; + + // Remote shares go through the direct/relay logic; our own shares (and + // browsers without JS or WebRTC) keep the plain relay link. + const handleDownload = (e) => { + if (!onDownload || canDelete || !rtcSupported()) return; + e.preventDefault(); + if (download) return; // already in progress + onDownload(share); + }; return (
@@ -75,13 +96,14 @@ function Share({ share, localShares, downloadHost, onRemove, onShowQR }) { - Download + {downloadLabel(download)}
); @@ -96,7 +118,7 @@ function Header() { ); } -function SharesList({ remoteShares, localShares, downloadHost, onRemove, onShowQR }) { +function SharesList({ remoteShares, localShares, downloadHost, downloads, onRemove, onShowQR, onDownload }) { return (

Shared files

@@ -109,8 +131,10 @@ function SharesList({ remoteShares, localShares, downloadHost, onRemove, onShowQ share={share} localShares={localShares} downloadHost={downloadHost} + downloads={downloads} onRemove={onRemove} onShowQR={onShowQR} + onDownload={onDownload} /> )) )} diff --git a/public/js/rtc.js b/public/js/rtc.js new file mode 100644 index 0000000..9d9a448 --- /dev/null +++ b/public/js/rtc.js @@ -0,0 +1,458 @@ +// --- Direct browser-to-browser transfer (WebRTC DataChannel) --------------- +// +// Devices behind the same public IP are usually on the same LAN. Instead of +// relaying the file through the server (up the uplink and back down), the +// receiver opens a DataChannel to the sender, signaled over the existing +// WebSocket. Only LAN host candidates are used (mDNS names or private +// addresses; no STUN, no TURN): the direct path can only succeed on the actual +// local network and no public address is disclosed to anybody. If the channel isn't open within RTC_CONNECT_TIMEOUT +// the download silently falls back to the relay (GET /share/:uuid), exactly +// as before. The server never sees the content: DTLS encrypts the channel +// end to end. +// +// Wire protocol on the channel (sender -> receiver): +// text {"size": } header +// binary ... file content, in order +// text {"done": true} end of file +// text {"error": true} sender gave up + +const RTC_CONNECT_TIMEOUT = 5000; // ms from offer to open channel (mDNS on Wi-Fi can take 1-2 s) +const RTC_CHUNK_SIZE = 64 * 1024; // safe SCTP message size everywhere +const RTC_HIGH_WATER = 4 * 1024 * 1024; // sender pauses above this bufferedAmount +const RTC_LOW_WATER = 1024 * 1024; // ...and resumes below this +const RTC_BLOB_MAX = 200 * 1024 * 1024; // in-memory sink limit without a service worker +const RTC_SENDER_LINGER = 60000; // ms a sender waits for the receiver to close + +const rtcSessions = {}; // session uuid => state (either role) +const rtcPendingOffers = {}; // share uuid => receiver state waiting for its session id + +function rtcSupported() { + return typeof RTCPeerConnection === 'function' && typeof MessageChannel === 'function'; +} + +// An address that can only be reached from the local network: an mDNS name +// (Chrome, Safari) or a private / link-local address (Firefox doesn't hide +// host candidates behind mDNS by default). Never a public address. +function rtcIsLanAddress(address) { + if (/\.local$/i.test(address)) return true; + const v4 = address.match(/^(\d+)\.(\d+)\.(\d+)\.(\d+)$/); + if (v4) { + const a = Number(v4[1]), b = Number(v4[2]); + return a === 10 || (a === 172 && b >= 16 && b <= 31) || (a === 192 && b === 168) || (a === 169 && b === 254); + } + // IPv6: unique local (fc00::/7) or link-local (fe80::/10) + return /^f[cd][0-9a-f]{2}:/i.test(address) || /^fe[89ab][0-9a-f]:/i.test(address); +} + +// "candidate:
typ ..." +// An empty string is the end-of-candidates marker. The server enforces the +// same rule; filtering here too keeps the LAN-only guarantee explicit. +function rtcIsLanCandidate(candidate) { + if (!candidate) return true; + const f = candidate.split(' '); + return f[6] === 'typ' && f[7] === 'host' && rtcIsLanAddress(f[4] || ''); +} + +function rtcSignal(state, msg) { + const ws = state.ws; + if (ws && ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(msg)); +} + +function rtcNewPeer(state) { + const pc = new RTCPeerConnection({ iceServers: [] }); + state.pc = pc; + state.pendingLocalIce = []; + state.pendingRemoteIce = []; + state.stats = { local: 0, localDropped: 0, remote: 0 }; + pc.onicecandidate = (e) => { + const c = e.candidate; + const candidate = c + ? { candidate: c.candidate, sdpMid: c.sdpMid, sdpMLineIndex: c.sdpMLineIndex, usernameFragment: c.usernameFragment } + : { candidate: '' }; + if (!rtcIsLanCandidate(candidate.candidate)) { + state.stats.localDropped++; + console.log('rtc: dropping non-LAN candidate: ' + candidate.candidate); + return; + } + if (c) state.stats.local++; + console.log('rtc ' + state.role + ': local candidate ' + (candidate.candidate || '(end)')); + if (state.session) { + rtcSignal(state, { type: 'rtc_ice', session: state.session, candidate }); + } else { + state.pendingLocalIce.push(candidate); + } + }; + pc.onconnectionstatechange = () => { + console.log('rtc ' + state.role + ': connection ' + pc.connectionState); + if (pc.connectionState === 'failed') rtcFail(state, 'ice-failed'); + }; + pc.oniceconnectionstatechange = () => console.log('rtc ' + state.role + ': ice ' + pc.iceConnectionState); + return pc; +} + +function rtcAddRemoteIce(state, candidate) { + if (!state.pc || !state.pc.remoteDescription) { + state.pendingRemoteIce.push(candidate); + return; + } + if (candidate.candidate) state.stats.remote++; + console.log('rtc ' + state.role + ': remote candidate ' + (candidate.candidate || '(end)') + ' mid=' + candidate.sdpMid + ' idx=' + candidate.sdpMLineIndex); + state.pc.addIceCandidate(candidate).catch((err) => console.warn('addIceCandidate failed', err)); +} + +function rtcFlushRemoteIce(state) { + const pending = state.pendingRemoteIce; + state.pendingRemoteIce = []; + pending.forEach((candidate) => rtcAddRemoteIce(state, candidate)); +} + +function rtcTeardown(state, reason) { + if (state.closed) return; + state.closed = true; + clearTimeout(state.timer); + if (state.share && rtcPendingOffers[state.share.uuid] === state) delete rtcPendingOffers[state.share.uuid]; + if (state.session) { + delete rtcSessions[state.session]; + if (!state.peerClosed) rtcSignal(state, { type: 'rtc_close', session: state.session, reason }); + } + const pc = state.pc; + const stats = state.stats || {}; + console.log('rtc ' + state.role + ' session ended: ' + reason + + (pc ? ' (ice ' + pc.iceConnectionState + ', gathering ' + pc.iceGatheringState + + ', local LAN candidates ' + stats.local + ', dropped ' + stats.localDropped + + ', remote ' + stats.remote + ', session ' + (state.session ? 'yes' : 'no') + ')' : '')); + try { if (state.dc) state.dc.close(); } catch (e) { /* already closed */ } + try { if (state.pc) state.pc.close(); } catch (e) { /* already closed */ } +} + +// The direct path is over for this transfer. For a receiver that hasn't +// finished, that means going back to the relay. +function rtcDumpPairs(pc, role) { + if (!pc || !pc.getStats) return; + pc.getStats().then((report) => { + const byId = {}; + report.forEach((r) => { byId[r.id] = r; }); + report.forEach((r) => { + if (r.type !== 'candidate-pair') return; + const l = byId[r.localCandidateId] || {}, m = byId[r.remoteCandidateId] || {}; + console.log('rtc ' + role + ': pair ' + r.state + ' ' + (l.address || l.ip) + ':' + l.port + '/' + l.protocol + + ' -> ' + (m.address || m.ip) + ':' + m.port + '/' + m.protocol + ' sent=' + r.requestsSent + ' recv=' + r.responsesReceived); + }); + }).catch(() => {}); +} + +function rtcFail(state, reason) { + if (state.closed || state.done) return; + rtcDumpPairs(state.pc, state.role); + rtcTeardown(state, reason); + if (state.role === 'receiver') { + if (state.sink) state.sink.abort(); + state.handlers.onFallback(reason); + } +} + +// --- Receiver ---------------------------------------------------------------- + +async function rtcDownload(share, ws, handlers) { + if (rtcPendingOffers[share.uuid]) return; // already in progress + const sink = await rtcCreateSink(share); + if (!sink) { + handlers.onFallback('no-sink'); + return; + } + const state = { + role: 'receiver', share, ws, handlers, sink, + session: null, received: 0, size: share.size, started: false, done: false, closed: false + }; + sink.oncancel = () => { + // The user cancelled the download in the browser UI: don't fall back. + state.done = true; + rtcTeardown(state, 'cancelled'); + handlers.onAbort(); + }; + const pc = rtcNewPeer(state); + const dc = pc.createDataChannel('file', { ordered: true }); + dc.binaryType = 'arraybuffer'; + state.dc = dc; + dc.onopen = () => { + clearTimeout(state.timer); + state.connected = true; + handlers.onConnected(); + }; + dc.onmessage = (e) => rtcReceiverMessage(state, e.data); + dc.onclose = () => rtcFail(state, 'channel-closed'); + dc.onerror = () => rtcFail(state, 'channel-error'); + + rtcPendingOffers[share.uuid] = state; + state.timer = setTimeout(() => rtcFail(state, 'timeout'), RTC_CONNECT_TIMEOUT); + try { + const offer = await pc.createOffer(); + await pc.setLocalDescription(offer); + if (state.closed) return; + rtcSignal(state, { type: 'rtc_offer', share: share.uuid, sdp: pc.localDescription.sdp }); + } catch (err) { + console.warn('rtc offer failed', err); + rtcFail(state, 'offer-error'); + } +} + +function rtcReceiverMessage(state, data) { + if (state.closed) return; + if (typeof data === 'string') { + let msg; + try { msg = JSON.parse(data); } catch (e) { return rtcFail(state, 'protocol'); } + if (!state.started && typeof msg.size === 'number') { + state.started = true; + state.size = msg.size; + // The filename comes from the share list (sanitized by the server), not + // from the peer. + Promise.resolve(state.sink.start(state.share.name, msg.size)) + .catch((err) => { console.warn('sink failed', err); rtcFail(state, 'sink-error'); }); + } else if (msg.done) { + if (state.received !== state.size) return rtcFail(state, 'incomplete'); + state.done = true; + state.sink.close(); + state.handlers.onDone(); + rtcTeardown(state, 'done'); + } else if (msg.error) { + rtcFail(state, 'sender-error'); + } + return; + } + if (!state.started) return rtcFail(state, 'protocol'); + state.received += data.byteLength; + if (state.received > state.size) return rtcFail(state, 'overflow'); + state.sink.write(data); + state.handlers.onProgress(state.received, state.size); +} + +// --- Receiver sinks ---------------------------------------------------------- +// +// A sink is { start(name, size), write(ArrayBuffer), close(), abort() } plus an +// `oncancel` callback. Preferred: a service worker streams the bytes straight +// into a regular browser download (no size limit, real progress bar). +// Fallback: buffer in memory and save a Blob, only for small files. Without +// either, the relay is used. + +async function rtcCreateSink(share) { + if (await rtcServiceWorkerReady()) return rtcStreamSink(); + if (typeof share.size === 'number' && share.size <= RTC_BLOB_MAX) return rtcBlobSink(); + return null; +} + +async function rtcServiceWorkerReady() { + if (!('serviceWorker' in navigator)) return false; + try { + if (navigator.serviceWorker.controller) return true; + const registration = await navigator.serviceWorker.getRegistration('/'); + if (!registration) return false; + // A freshly registered worker claims open pages on activation; give it a moment. + await Promise.race([navigator.serviceWorker.ready, new Promise((r) => setTimeout(r, 1000))]); + return !!navigator.serviceWorker.controller; + } catch (e) { + return false; + } +} + +function rtcStreamSink() { + const id = generateUUID(); + const queue = []; + let port = null; + let ready = false; + let iframe = null; + const sink = { oncancel: null }; + + const post = (msg, transfer) => port.postMessage(msg, transfer || []); + const removeIframe = () => { if (iframe) { iframe.remove(); iframe = null; } }; + + sink.start = (name, size) => new Promise((resolve, reject) => { + const channel = new MessageChannel(); + port = channel.port1; + const timer = setTimeout(() => reject(new Error('service worker did not answer')), 2000); + port.onmessage = (e) => { + const msg = e.data || {}; + if (msg.type === 'dl-ready') { + clearTimeout(timer); + ready = true; + while (queue.length) { const buf = queue.shift(); post(buf, [buf]); } + // Navigating a hidden iframe to the worker-served URL starts a normal + // browser download without leaving the page. + iframe = document.createElement('iframe'); + iframe.hidden = true; + iframe.src = '/dl/' + id; + document.body.appendChild(iframe); + resolve(); + } else if (msg.type === 'dl-cancel') { + if (sink.oncancel) sink.oncancel(); + } + }; + navigator.serviceWorker.controller.postMessage({ type: 'dl-open', id, name, size }, [channel.port2]); + }); + sink.write = (buf) => { + if (ready) post(buf, [buf]); + else queue.push(buf); + }; + sink.close = () => { + post({ type: 'dl-end' }); + setTimeout(removeIframe, RTC_SENDER_LINGER); + }; + sink.abort = () => { + if (port) post({ type: 'dl-abort' }); + removeIframe(); + }; + return sink; +} + +function rtcBlobSink() { + const chunks = []; + let name = 'download'; + return { + oncancel: null, + start(n) { name = n; }, + write(buf) { chunks.push(buf); }, + close() { + const url = URL.createObjectURL(new Blob(chunks)); + const a = document.createElement('a'); + a.href = url; + a.download = name; + document.body.appendChild(a); + a.click(); + a.remove(); + setTimeout(() => URL.revokeObjectURL(url), RTC_SENDER_LINGER); + }, + abort() { chunks.length = 0; } + }; +} + +// --- Sender ------------------------------------------------------------------ + +async function rtcHandleOffer(msg, ws, localShares) { + const share = localShares[msg.share]; + const state = { role: 'sender', share, ws, session: msg.session, closed: false }; + if (!share) { + rtcSignal(state, { type: 'rtc_close', session: msg.session, reason: 'unknown-share' }); + return; + } + rtcSessions[msg.session] = state; + const pc = rtcNewPeer(state); + pc.ondatachannel = (e) => { + const dc = e.channel; + dc.binaryType = 'arraybuffer'; + dc.bufferedAmountLowThreshold = RTC_LOW_WATER; + state.dc = dc; + dc.onclose = () => rtcTeardown(state, 'channel-closed'); + dc.onerror = () => rtcTeardown(state, 'channel-error'); + // Chrome may hand us a channel that is already open: onopen won't fire then. + if (dc.readyState === 'open') rtcPump(state); + else dc.onopen = () => rtcPump(state); + }; + state.timer = setTimeout(() => { + if (!state.dc || state.dc.readyState !== 'open') rtcTeardown(state, 'timeout'); + }, RTC_CONNECT_TIMEOUT * 2); + try { + await pc.setRemoteDescription({ type: 'offer', sdp: msg.sdp }); + rtcFlushRemoteIce(state); + const answer = await pc.createAnswer(); + await pc.setLocalDescription(answer); + if (state.closed) return; + rtcSignal(state, { type: 'rtc_answer', session: msg.session, sdp: pc.localDescription.sdp }); + state.pendingLocalIce.forEach((candidate) => rtcSignal(state, { type: 'rtc_ice', session: state.session, candidate })); + state.pendingLocalIce = []; + } catch (err) { + console.warn('rtc answer failed', err); + rtcTeardown(state, 'answer-error'); + } +} + +function rtcBufferedLow(dc) { + return new Promise((resolve) => { + if (dc.readyState !== 'open' || dc.bufferedAmount <= RTC_LOW_WATER) return resolve(); + const done = () => { + dc.removeEventListener('bufferedamountlow', done); + dc.removeEventListener('close', done); + resolve(); + }; + dc.addEventListener('bufferedamountlow', done); + dc.addEventListener('close', done); + }); +} + +async function rtcPump(state) { + const { dc, share } = state; + clearTimeout(state.timer); + const source = share.file || new Blob([share.content || ''], { type: 'text/plain' }); + try { + dc.send(JSON.stringify({ size: source.size })); + let pos = 0; + while (pos < source.size) { + if (state.closed || dc.readyState !== 'open') return; + if (dc.bufferedAmount > RTC_HIGH_WATER) { + await rtcBufferedLow(dc); + continue; + } + const buf = await source.slice(pos, pos + RTC_CHUNK_SIZE).arrayBuffer(); + if (state.closed || dc.readyState !== 'open') return; + dc.send(buf); + pos += buf.byteLength; + } + dc.send(JSON.stringify({ done: true })); + // The receiver closes the session once it has everything. + state.timer = setTimeout(() => rtcTeardown(state, 'linger-timeout'), RTC_SENDER_LINGER); + } catch (err) { + console.warn('rtc send failed', err); + try { dc.send(JSON.stringify({ error: true })); } catch (e) { /* channel gone */ } + rtcTeardown(state, 'send-error'); + } +} + +// --- Signaling messages from the server ------------------------------------- + +function rtcHandleMessage(msg, ws, localShares) { + switch (msg.type) { + case 'rtc_session': { + const state = rtcPendingOffers[msg.share]; + if (!state) return; + delete rtcPendingOffers[msg.share]; + state.session = msg.session; + rtcSessions[msg.session] = state; + state.pendingLocalIce.forEach((candidate) => rtcSignal(state, { type: 'rtc_ice', session: msg.session, candidate })); + state.pendingLocalIce = []; + break; + } + case 'rtc_offer': + rtcHandleOffer(msg, ws, localShares); + break; + case 'rtc_answer': { + const state = rtcSessions[msg.session]; + if (!state || state.role !== 'receiver') return; + state.pc.setRemoteDescription({ type: 'answer', sdp: msg.sdp }) + .then(() => rtcFlushRemoteIce(state)) + .catch((err) => { console.warn('rtc setRemoteDescription failed', err); rtcFail(state, 'answer-error'); }); + break; + } + case 'rtc_ice': { + const state = rtcSessions[msg.session]; + if (state) rtcAddRemoteIce(state, msg.candidate); + break; + } + case 'rtc_close': { + const state = rtcSessions[msg.session]; + if (!state) return; + state.peerClosed = true; + if (state.done) rtcTeardown(state, 'peer-closed'); + else rtcFail(state, 'peer-closed'); + break; + } + default: + break; + } +} + +// The WebSocket dropped: handshakes in flight can't complete. Established +// channels don't need the socket and carry on. +function rtcSocketClosed() { + Object.values(rtcPendingOffers).forEach((state) => rtcFail(state, 'socket-closed')); + Object.values(rtcSessions).forEach((state) => { + if (!state.connected && state.role === 'receiver') rtcFail(state, 'socket-closed'); + }); +} diff --git a/public/js/utils.js b/public/js/utils.js index 70d01c2..eb22592 100644 --- a/public/js/utils.js +++ b/public/js/utils.js @@ -1,10 +1,26 @@ +// Share uuids are capability tokens (anyone holding /share/ can fetch +// the file), so they must come from a cryptographic source, not Math.random(). function generateUUID() { - let d = new Date().getTime(); - return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function (c) { - const r = (d + Math.random() * 16) % 16 | 0; - d = Math.floor(d / 16); - return (c === 'x' ? r : (r & 0x3 | 0x8)).toString(16); - }); + if (crypto.randomUUID) return crypto.randomUUID(); + // Older browsers / insecure contexts: RFC 4122 v4 from getRandomValues. + const bytes = crypto.getRandomValues(new Uint8Array(16)); + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + const hex = Array.from(bytes, (b) => b.toString(16).padStart(2, '0')).join(''); + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`; +} + +// Start a download of `url` (an attachment response) without navigating: +// window.location.assign() counts as a navigation and makes some browsers +// (Firefox, Safari) drop the page's WebSocket. +function relayDownload(url) { + const a = document.createElement('a'); + a.href = url; + a.download = ''; + a.style.display = 'none'; + document.body.appendChild(a); + a.click(); + setTimeout(() => a.remove(), 1000); } function ellipseAt(str, length) { diff --git a/public/sw.js b/public/sw.js new file mode 100644 index 0000000..d5acb14 --- /dev/null +++ b/public/sw.js @@ -0,0 +1,61 @@ +// Service worker: turns a stream of chunks pushed by the page (received over a +// WebRTC DataChannel) into a regular browser download, so that a direct +// transfer never has to hold the whole file in memory. +// +// The page sends {type: 'dl-open', id, name, size} with a MessagePort, then +// pushes ArrayBuffers on that port, then {type: 'dl-end'} (or 'dl-abort'). +// Navigating to /dl/ is answered with a streaming attachment response. + +const streams = new Map(); + +self.addEventListener('install', () => self.skipWaiting()); +self.addEventListener('activate', (event) => event.waitUntil(self.clients.claim())); + +self.addEventListener('message', (event) => { + const data = event.data; + if (!data || data.type !== 'dl-open' || !event.ports[0]) return; + const port = event.ports[0]; + let controller = null; + const stream = new ReadableStream({ + start(c) { controller = c; }, + cancel() { + streams.delete(data.id); + port.postMessage({ type: 'dl-cancel' }); + } + }); + port.onmessage = (e) => { + const msg = e.data; + try { + if (msg instanceof ArrayBuffer) { + controller.enqueue(new Uint8Array(msg)); + } else if (msg && msg.type === 'dl-end') { + controller.close(); + } else if (msg && msg.type === 'dl-abort') { + streams.delete(data.id); + controller.error(new Error('transfer aborted')); + } + } catch (err) { + // Stream already closed or errored (e.g. download cancelled). + } + }; + streams.set(data.id, { name: String(data.name || 'download'), size: data.size, stream }); + port.postMessage({ type: 'dl-ready' }); +}); + +self.addEventListener('fetch', (event) => { + const url = new URL(event.request.url); + const match = url.pathname.match(/^\/dl\/([a-f0-9-]{36})$/i); + if (!match) return; + const entry = streams.get(match[1]); + if (!entry) return; + streams.delete(match[1]); + const asciiName = entry.name.replace(/[^\x20-\x7e]/g, '_').replace(/["\\]/g, '_'); + const headers = { + 'Content-Type': 'application/octet-stream', + 'Content-Disposition': `attachment; filename="${asciiName}"; filename*=UTF-8''${encodeURIComponent(entry.name)}`, + 'X-Content-Type-Options': 'nosniff', + 'Cache-Control': 'no-store' + }; + if (typeof entry.size === 'number') headers['Content-Length'] = String(entry.size); + event.respondWith(new Response(entry.stream, { headers })); +}); diff --git a/spec/app_spec.rb b/spec/app_spec.rb index 3716bcb..965268d 100644 --- a/spec/app_spec.rb +++ b/spec/app_spec.rb @@ -123,4 +123,29 @@ def ws_headers(connection: 'Upgrade') get "/share/#{SecureRandom.uuid}" expect(last_response.status).to eq(404) end + + # The client IP is the whole authorization model (namespaces are keyed by + # it), so pin down how Rack resolves it behind the nginx-proxy container. + describe "client IP resolution" do + let(:namespace) { registry.context_for("198.51.100.7").namespace_for(:default) } + + it "uses X-Forwarded-For when the request comes through the proxy" do + share + get "/g", {}, 'REMOTE_ADDR' => '172.18.0.2', 'HTTP_X_FORWARDED_FOR' => '198.51.100.7' + expect(last_response).to be_ok + expect(last_response.body).to eq("hello") + end + + it "ignores X-Forwarded-For sent directly by a public address" do + share + get "/g", {}, 'REMOTE_ADDR' => '203.0.113.5', 'HTTP_X_FORWARDED_FOR' => '198.51.100.7' + expect(last_response.status).to eq(404) + end + + it "keeps the last untrusted hop when several proxies are chained" do + share + get "/g", {}, 'REMOTE_ADDR' => '172.18.0.2', 'HTTP_X_FORWARDED_FOR' => '10.0.0.9, 198.51.100.7, 172.18.0.3' + expect(last_response).to be_ok + end + end end diff --git a/spec/ws_client_spec.rb b/spec/ws_client_spec.rb index 1022055..a3ee620 100644 --- a/spec/ws_client_spec.rb +++ b/spec/ws_client_spec.rb @@ -157,3 +157,132 @@ def chunk_msg(stream_uuid, data, close: false) end end end + +RSpec.describe DLCenter::WSClient, "WebRTC signaling" do + let(:namespace) { DLCenter::Namespace.new(:default) } + let(:sender_ws) { FakeWS.new } + let(:receiver_ws) { FakeWS.new } + let(:sender) { DLCenter::WSClient.new(namespace, sender_ws) } + let(:receiver) { DLCenter::WSClient.new(namespace, receiver_ws) } + let(:share) { DLCenter::Share.new(sender, name: "file", size: 3) } + let(:mdns) { "candidate:1 1 udp 2113937151 3f0a1c9e-2b7d-4f5a-9c1e-8a2b3c4d5e6f.local 54321 typ host generation 0" } + let(:private_ip) { "candidate:2 1 udp 2113937151 192.168.1.10 54321 typ host generation 0" } + let(:link_local_v6) { "candidate:4 1 udp 2113937151 fe80::1c2a:3b4c:5d6e:7f80 54321 typ host generation 0" } + let(:public_ip) { "candidate:5 1 udp 2113937151 203.0.113.5 54321 typ host generation 0" } + let(:srflx) { "candidate:3 1 udp 1677729535 192.168.1.10 54321 typ srflx raddr 0.0.0.0 rport 0" } + + before do + namespace.add_client(sender) + namespace.add_client(receiver) + sender.add_share(share) + end + + def offer! + receiver_ws.message(type: 'rtc_offer', share: share.uuid, sdp: "v=0\r\n") + receiver_ws.sent.find { |m| m[:type] == 'rtc_session' } + end + + it "opens a session toward the share owner without exposing client identities" do + session = offer! + expect(session).to include(share: share.uuid) + offer = sender_ws.sent.find { |m| m[:type] == 'rtc_offer' } + expect(offer).to eq(type: 'rtc_offer', session: session[:session], share: share.uuid, sdp: "v=0\r\n") + expect(sender.rtc_sessions.keys).to eq([session[:session]]) + expect(receiver.rtc_sessions.keys).to eq([session[:session]]) + end + + it "relays the answer and mDNS candidates by session" do + session = offer![:session] + sender_ws.message(type: 'rtc_answer', session: session, sdp: "v=0\r\nanswer") + sender_ws.message(type: 'rtc_ice', session: session, + candidate: { candidate: mdns, sdpMid: '0', sdpMLineIndex: 0, extra: 'dropped' }) + receiver_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: '' }) + expect(receiver_ws.sent).to include(type: 'rtc_answer', session: session, sdp: "v=0\r\nanswer") + expect(receiver_ws.sent).to include(type: 'rtc_ice', session: session, + candidate: { candidate: mdns, sdpMid: '0', sdpMLineIndex: 0 }) + expect(sender_ws.sent).to include(type: 'rtc_ice', session: session, candidate: { candidate: '' }) + end + + it "relays private and link-local host candidates (Firefox doesn't use mDNS)" do + session = offer![:session] + sender_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: private_ip }) + sender_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: link_local_v6 }) + relayed = receiver_ws.sent.select { |m| m[:type] == 'rtc_ice' }.map { |m| m[:candidate][:candidate] } + expect(relayed).to eq([private_ip, link_local_v6]) + end + + it "drops public, server-reflexive and malformed candidates" do + session = offer![:session] + sender_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: public_ip }) + sender_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: srflx }) + sender_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: "candidate:1 1 udp 1 not-an-ip 1 typ host" }) + sender_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: "garbage" }) + expect(receiver_ws.sent.select { |m| m[:type] == 'rtc_ice' }).to be_empty + end + + it "scrubs non-LAN candidate lines embedded in the SDP" do + receiver_ws.message(type: 'rtc_offer', share: share.uuid, + sdp: "v=0\r\na=#{public_ip}\r\na=#{srflx}\r\na=#{mdns}\r\na=#{private_ip}\r\na=end-of-candidates\r\n") + offer = sender_ws.sent.find { |m| m[:type] == 'rtc_offer' } + expect(offer[:sdp]).to eq("v=0\r\na=#{mdns}\r\na=#{private_ip}\r\na=end-of-candidates\r\n") + end + + it "refuses to signal toward a share from another namespace" do + other = DLCenter::Namespace.new(:other) + stranger_ws = FakeWS.new + stranger = DLCenter::WSClient.new(other, stranger_ws) + other.add_client(stranger) + stranger_ws.message(type: 'rtc_offer', share: share.uuid, sdp: "v=0\r\n") + expect(stranger_ws.sent.map { |m| m[:type] }).not_to include('rtc_session') + expect(sender_ws.sent.map { |m| m[:type] }).not_to include('rtc_offer') + expect(sender.rtc_sessions).to be_empty + end + + it "refuses to open a session toward oneself or a non-browser sender" do + sender_ws.message(type: 'rtc_offer', share: share.uuid, sdp: "v=0\r\n") + io_client = DLCenter::IOClient.new(namespace, StringIO.new, StringIO.new, filename: 'curl.bin') + namespace.add_client(io_client) + receiver_ws.message(type: 'rtc_offer', share: io_client.shares.keys.first, sdp: "v=0\r\n") + expect(sender_ws.sent.map { |m| m[:type] }).not_to include('rtc_session', 'rtc_offer') + expect(receiver_ws.sent.map { |m| m[:type] }).not_to include('rtc_session') + end + + it "ignores signaling for an unknown session" do + stranger_ws = FakeWS.new + stranger = DLCenter::WSClient.new(namespace, stranger_ws) + namespace.add_client(stranger) + session = offer![:session] + stranger_ws.message(type: 'rtc_answer', session: session, sdp: "v=0\r\n") + stranger_ws.message(type: 'rtc_ice', session: session, candidate: { candidate: mdns }) + expect(receiver_ws.sent.map { |m| m[:type] }).not_to include('rtc_answer', 'rtc_ice') + end + + it "rejects oversized SDP" do + receiver_ws.message(type: 'rtc_offer', share: share.uuid, sdp: "a" * (DLCenter::MAX_RTC_SDP_LENGTH + 1)) + expect(receiver_ws.sent.map { |m| m[:type] }).not_to include('rtc_session') + end + + it "caps the number of concurrent sessions per client" do + DLCenter::MAX_RTC_SESSIONS_PER_CLIENT.times { offer! } + expect(receiver.rtc_sessions.size).to eq(DLCenter::MAX_RTC_SESSIONS_PER_CLIENT) + receiver_ws.sent.clear + expect(offer!).to be_nil + end + + it "closes the session on both sides and tells the peer" do + session = offer![:session] + receiver_ws.message(type: 'rtc_close', session: session, reason: 'done') + expect(sender_ws.sent).to include(type: 'rtc_close', session: session) + expect(sender.rtc_sessions).to be_empty + expect(receiver.rtc_sessions).to be_empty + # Nothing echoes back to the side that closed + expect(receiver_ws.sent.map { |m| m[:type] }).not_to include('rtc_close') + end + + it "tells peers when a client disconnects mid-handshake" do + session = offer![:session] + sender_ws.close + expect(receiver_ws.sent).to include(type: 'rtc_close', session: session) + expect(receiver.rtc_sessions).to be_empty + end +end