restored/websocket.inc
← All source files · Download this file
11,856 bytes · SHA256 628fce98c58443d077ab3e964a1282371a1421c326ed14cb47292e7825f901db
// Staged WebSocket operations over the Stage1 owned transport buffers.
bool vScript::websocketCommand(Frame& frame,const CodeLine& c) {
const auto& a=c.arg; const auto& s=c.source; const auto& sub=a[1].text;
auto val=[&](std::size_t i) { return value(frame,a[i],s); };
if(!options.allowNetwork) fail(s,"websocket operations require --allow-network");
if(++socketCalls>4096) failLimit(s,"socket call limit (4096)");
bool stepping=sub=="step";
if(sub!="abort") {
auto& out=lookup(frame,a[2],s); if(out.array) fail(s,"websocket metadata destination must be scalar or JSON");
if(stepping) { auto& data=lookup(frame,a[3],s); if(data.array||data.structured) fail(s,"websocket data destination must be scalar"); if(a[2].text==a[3].text) fail(s,"websocket destinations must differ"); }
std::size_t total=frame.extraBytes+65536+32768;
for(const auto& item:frame.vars) total+=variableSize(item.second)+item.first.capacity();
if(total>frameLimit) failLimit(s,"websocket frame reservation limit");
}
auto key=val(sub=="abort" ? 2 : stepping ? 4 : 3); auto it=sockets.find(key);
if(it==sockets.end()) fail(s,"invalid or closed socket handle");
auto p=it->second;
if(p->listener||p->datagram) fail(s,"websocket requires TCP connection");
if(p->handshaking && SocketClock::now()>=p->handshakeDeadline) {
sockets.erase(key); fail(s,"socket TLS handshake timeout");
}
auto bytesRoom=[&](std::size_t n) { if(n>16*valueLimit-socketBytes) failLimit(s,"socket cumulative byte limit (16 MiB)"); };
auto storage=[&](std::size_t extra) {
std::size_t n=0; for(const auto& e:sockets) n+=e.second->buffer.capacity()+e.second->pendingRead.capacity()+e.second->pendingWrite.capacity()+(e.second->ws ? e.second->ws->storage() : 0);
if(n+extra>valueLimit) failLimit(s,"socket buffer storage limit (1 MiB)");
};
auto osError=[&](const char* operation) {
int error=errno;
if(error==ENOMEM || error==ENOBUFS || error==EMFILE || error==ENFILE) {
for(auto& entry:sockets) { if(entry.second->fd>=0) ::close(entry.second->fd); entry.second->fd=-1; }
sockets.clear(); failLimit(s,"websocket OS resource limit");
}
fail(s,std::string("websocket transport ")+operation+" failed: "+std::strerror(error));
};
auto reserveTime=[&](std::size_t n) { if(n>60000-socketTime) failLimit(s,"socket cumulative timeout reservation limit (60000 ms)"); socketTime+=n; };
if(sub=="abort") { if(!p->ws) fail(s,"websocket handle required"); sockets.erase(key); return true; }
std::size_t accepted=0,countWire=0,completed=0; std::string data,type,status="progress";
auto result=[&]() {
auto& w=*p->ws;
bool pending=!p->pendingWrite.empty()||!w.queue.empty();
std::string want;
if(status!="closed") want=p->tls && !p->pendingRead.empty() && p->readWant==POLLOUT ? "w" : pending ? (p->want==POLLIN && !p->pendingWrite.empty() ? "r" : "w") : "r";
Json out={{"status",status},{"want",want},{"accepted",accepted},{"count",countWire},{"completed",completed},{"pending",pending},{"type",type},{"code",w.closeCode}};
if(stepping) assign(frame,a[3],data,s);
putJson(frame,a[2],std::move(out),s);
};
if(sub=="client"||sub=="server") {
if(p->ws) fail(s,"websocket already claimed");
if(p->handshaking || !p->buffer.empty() || !p->pendingRead.empty() || !p->pendingWrite.empty() || p->eof) fail(s,"websocket requires idle established stream");
if(sub=="server" && !options.allowListen) fail(s,"websocket server requires --allow-listen");
auto config=jsonValue(frame,a[4],s);
if(!config.is_object()||config.size()!=4) fail(s,"websocket config requires host/path/origin/protocol");
for(const auto* name:{"host","path","origin","protocol"}) if(!config.contains(name)||!config[name].is_string()) fail(s,"websocket config fields must be strings");
auto w=std::make_unique<SocketWebSocket>(); w->client=sub=="client";
w->host=config["host"].get<std::string>(); w->path=config["path"].get<std::string>(); w->origin=config["origin"].get<std::string>(); w->protocol=config["protocol"].get<std::string>();
for(const auto* field:{&w->host,&w->path,&w->origin,&w->protocol}) wsSafeField(*field,s);
if(w->host.empty()||w->host.find_first_of("/\\@?#")!=std::string::npos || w->path.empty()||w->path[0]!='/'||w->path.find('#')!=std::string::npos || (!w->protocol.empty()&&!wsToken(w->protocol))) fail(s,"websocket invalid host/path/protocol");
auto ms=static_cast<std::size_t>(integer(val(5),1,30000,s));
storage(32768); bytesRoom(8192);
if(w->client) {
unsigned char random[16]; if(RAND_bytes(random,16)!=1) fail(s,"websocket random key failed");
auto nonce=wsBase64(random,16); w->accept=wsAccept(nonce,s);
SocketWebSocket::Output request; request.wire="GET "+w->path+" HTTP/1.1\r\nHost: "+w->host+"\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: "+nonce+"\r\n";
if(!w->origin.empty()) request.wire+="Origin: "+w->origin+"\r\n";
if(!w->protocol.empty()) request.wire+="Sec-WebSocket-Protocol: "+w->protocol+"\r\n";
request.wire+="\r\n"; socketBytes+=request.wire.size(); w->queue.push_back(std::move(request)); w->phase=SocketWebSocket::sending;
}
reserveTime(ms+1000); w->deadline=SocketClock::now()+std::chrono::milliseconds(ms);
p->ws=std::move(w); status="handshake"; result(); return true;
}
if(!p->ws) fail(s,"websocket handle required");
auto& w=*p->ws;
if(w.timed()&&SocketClock::now()>=w.deadline) { sockets.erase(key); fail(s,"websocket deadline expired"); }
if(sub=="send"||sub=="close") {
if(w.phase!=SocketWebSocket::open) fail(s,"websocket is not open");
std::string payload; unsigned opcode;
if(sub=="send") {
auto kind=val(4); payload=val(5);
if(kind!="text"&&kind!="binary"&&kind!="ping") fail(s,"websocket invalid send type");
opcode=kind=="text" ? 1U : kind=="binary" ? 2U : 9U;
if(payload.size()>(opcode==9 ? 125U : 65536U)) fail(s,"websocket send size limit");
if(opcode==1&&!wsUTF8(payload)) fail(s,"websocket invalid UTF8 text");
if(opcode!=9&&w.applicationPending) fail(s,"websocket application output pending");
} else {
auto code=static_cast<unsigned>(integer(val(4),1000,4999,s)); auto reason=val(5);
if(!wsCode(code)||reason.size()>123||!wsUTF8(reason)) fail(s,"websocket invalid close code/reason");
payload.push_back(static_cast<char>(code>>8)); payload.push_back(static_cast<char>(code)); payload+=reason; opcode=8;
}
if(w.queue.size()>=8) fail(s,"websocket output queue limit");
storage(2*(payload.size()+64)+4096); bytesRoom(payload.size()+14);
std::size_t ms=sub=="close" ? static_cast<std::size_t>(integer(val(6),1,30000,s)) : 0;
if(ms>60000-socketTime) failLimit(s,"socket cumulative timeout reservation limit (60000 ms)");
w.enqueue(opcode,payload,s); socketBytes+=w.queue.back().wire.size(); accepted=payload.size();
if(opcode==1||opcode==2) w.applicationPending=true;
if(sub=="close") { reserveTime(ms); w.phase=SocketWebSocket::closing; w.deadline=SocketClock::now()+std::chrono::milliseconds(ms); status="closing"; }
result(); return true;
}
// Reserve temporary copies (frame decoding, reassembly, result and control).
storage(2*(w.input.size()+w.message.size()+8192)+16384); bytesRoom(16384);
try {
bool readExclusive=p->tls&&!p->pendingRead.empty()&&p->readWant==POLLOUT;
if(!readExclusive && (!p->pendingWrite.empty()||!w.queue.empty())) {
if(p->pendingWrite.empty()) {
auto& q=w.queue.front(); p->pendingWrite.swap(q.wire); w.activePayload=q.payload; w.activeOpcode=q.opcode; w.queue.erase(w.queue.begin());
}
auto n=p->tls ? p->tls->step(2,p->pendingWrite.data(),static_cast<int>(p->pendingWrite.size()),p->want,s) : ::send(p->fd,p->pendingWrite.data(),p->pendingWrite.size(),MSG_NOSIGNAL);
if(n<0) { if(!p->tls&&!socketWouldBlock()) osError("send"); if(!p->tls) p->want=POLLOUT; status="wouldblock"; }
else if(n==0) fail(s,"websocket transport write made no progress");
else {
countWire=static_cast<std::size_t>(n); p->pendingWrite=std::string(p->pendingWrite.data()+n,p->pendingWrite.size()-countWire);
if(p->pendingWrite.empty()) {
completed=w.activePayload;
if(w.activeOpcode==1||w.activeOpcode==2) w.applicationPending=false;
if(w.activeOpcode==8) w.closeSent=true;
w.activePayload=0; w.activeOpcode=0; p->want=p->readWant;
if(w.phase==SocketWebSocket::sending) { w.phase=w.client ? SocketWebSocket::receiving : SocketWebSocket::open; status=w.client ? "handshake" : "open"; }
}
}
} else {
bool handled=false;
auto parse=[&]() {
if(w.phase==SocketWebSocket::receiving) { if(w.upgrade(s)) { status=w.phase==SocketWebSocket::open ? "open" : "handshake"; handled=true; if(!w.client) { socketBytes+=w.queue.back().wire.size(); } } }
else if(w.phase==SocketWebSocket::open||w.phase==SocketWebSocket::closing) { status=w.parse(data,type,s); handled=!status.empty(); }
};
if(w.parseReady&&!readExclusive) parse();
if(!handled) {
bool attempt=true;
if(p->tls&&p->pendingRead.empty()&&SSL_pending(p->tls->ssl)==0) {
pollfd probe{p->fd,POLLIN,0}; int rc=::poll(&probe,1,0); if(rc<0&&errno!=EINTR) osError("poll"); attempt=rc>0;
}
ssize_t n=-1;
if(attempt) {
if(p->pendingRead.empty()) { p->pendingRead=std::string(8192,'\0'); p->readMax=8192; }
n=p->tls ? p->tls->step(1,p->pendingRead.data(),8192,p->want,s) : ::recv(p->fd,p->pendingRead.data(),8192,0);
if(p->tls) p->readWant=p->want;
if(n<0&&!p->tls&&!socketWouldBlock()) osError("receive");
if(n>0) {
socketBytes+=static_cast<std::size_t>(n);
if(w.input.size()+static_cast<std::size_t>(n)>73742) fail(s,"websocket input buffer limit");
std::string fresh=w.input; fresh.append(p->pendingRead.data(),static_cast<std::size_t>(n)); w.input.swap(fresh); w.parseReady=true;
}
if(n>=0||!p->tls) std::string().swap(p->pendingRead);
}
if(n==0) fail(s,"websocket truncated EOF before close exchange");
if(n>0) parse();
if(!handled) status=n<0 ? "wouldblock" : "progress";
}
if(status=="ping") { w.enqueue(10,data,s); socketBytes+=w.queue.back().wire.size(); }
if(status=="closing" && w.closeReceived) {
if(w.phase!=SocketWebSocket::closing) {
w.phase=SocketWebSocket::closing; w.deadline=SocketClock::now()+std::chrono::milliseconds(1000);
w.enqueue(8,data,s); socketBytes+=w.queue.back().wire.size();
}
data=data.empty() ? "" : data.substr(2);
}
}
if(w.timed()&&SocketClock::now()>=w.deadline) fail(s,"websocket deadline expired");
if(w.closeSent&&w.closeReceived) { status="closed"; sockets.erase(key); }
} catch(...) { sockets.erase(key); throw; }
result(); return true;
}