restored/socket_foundations.inc
← All source files · Download this file
17,613 bytes · SHA256 13770d123d3130d1746bd64eb59367e5832d779eab9c52a3fd83887c8c89b1d6
// Stage 1 event-loop operations. No blocking transport calls in this module.
static bool foundationOperation(const std::string& sub) {
return sub=="wait" || sub=="tryaccept" || sub=="tryrecv" || sub=="trysend" || sub=="handshake" || sub=="tlslisten";
}
static void foundationValidate(const CodeLine& c) {
count(c,2,7); const auto& sub=c.arg[1].text;
bool udp=c.arg[0].text=="udp";
if(udp) {
if(sub=="bind" || sub=="endpoint") count(c,5,5);
else if(sub=="sendto") count(c,7,7);
else if(sub=="recvfrom") count(c,6,6);
else if(sub=="close") { count(c,3,3); return; }
else fail(c.source,"unknown udp operation");
} else if(sub=="wait" || sub=="trysend") count(c,5,5);
else if(sub=="handshake") count(c,4,4);
else count(c,6,6);
destination(c.arg[2],c.source);
if(sub=="tryrecv" || sub=="tryaccept" || sub=="recvfrom" || sub=="endpoint") destination(c.arg[3],c.source);
}
static bool socketWouldBlock() { return errno==EAGAIN || errno==EWOULDBLOCK || errno==EINTR; }
static std::pair<std::string,unsigned> socketPeer(const sockaddr_storage& addr,const Source& s) {
char text[INET6_ADDRSTRLEN]{}; unsigned port;
if(addr.ss_family==AF_INET) {
auto* x=reinterpret_cast<const sockaddr_in*>(&addr);
if(!inet_ntop(AF_INET,&x->sin_addr,text,sizeof(text))) fail(s,"socket address conversion failed");
port=ntohs(x->sin_port);
} else if(addr.ss_family==AF_INET6) {
auto* x=reinterpret_cast<const sockaddr_in6*>(&addr);
if(!inet_ntop(AF_INET6,&x->sin6_addr,text,sizeof(text))) fail(s,"socket address conversion failed");
port=ntohs(x->sin6_port);
} else fail(s,"socket unsupported address family");
return {text,port};
}
bool vScript::foundationCommand(Frame& frame,const CodeLine& c) {
const auto& a=c.arg; const auto& s=c.source; const auto& sub=a[1].text;
bool udp=a[0].text=="udp";
std::shared_ptr<Socket> creating; // Also release not-yet-published FDs on fatal OS errors.
auto val=[&](std::size_t i) { return value(frame,a[i],s); };
auto put=[&](std::size_t i,const std::string& x) { assign(frame,a[i],x,s); };
auto result=[&](Json j) { putJson(frame,a[2],std::move(j),s); };
auto reserve=[&](std::size_t bytes,bool second=false,bool scalarFirst=false) {
auto& first=lookup(frame,a[2],s);
if(first.array || (scalarFirst && first.structured)) fail(s,"socket destination must be scalar");
if(second) {
auto& other=lookup(frame,a[3],s);
if(other.array || other.structured) fail(s,"socket destination must be scalar");
if(a[2].text==a[3].text) fail(s,"socket destinations must differ");
}
std::size_t total=frame.extraBytes+bytes+16384;
for(const auto& entry:frame.vars) total+=variableSize(entry.second)+entry.first.capacity();
if(total>frameLimit) failLimit(s,"socket frame reservation limit");
};
if(!options.allowNetwork) fail(s,"socket operations require --allow-network (not a sandbox)");
if(++socketCalls>4096) failLimit(s,"socket call limit (4096)");
auto osError=[&](const char* op) {
int error=errno;
if(error==ENOMEM || error==ENOBUFS || error==EMFILE || error==ENFILE) {
if(creating && creating->fd>=0) { ::close(creating->fd); creating->fd=-1; }
// shared_ptr temporaries may retain Socket; close FDs explicitly first.
for(auto& entry:sockets) { if(entry.second->fd>=0) ::close(entry.second->fd); entry.second->fd=-1; }
sockets.clear(); failLimit(s,"socket OS resource limit: "+std::string(op));
}
fail(s,"socket "+std::string(op)+": "+std::strerror(error));
};
auto get=[&](const std::string& key) {
auto it=sockets.find(key); if(it==sockets.end()) fail(s,"invalid or closed socket handle");
return it->second;
};
auto room=[&]() { if(sockets.size()>=32) failLimit(s,"socket handle limit (32)"); };
auto timeRoom=[&](std::size_t ms) { if(ms>60000-socketTime) failLimit(s,"socket cumulative timeout reservation limit (60000 ms)"); };
auto bytesRoom=[&](std::size_t bytes) { if(bytes>16*valueLimit-socketBytes) failLimit(s,"socket cumulative byte limit (16 MiB)"); };
auto storage=[&](std::size_t bytes) {
std::size_t retained=0;
for(const auto& entry:sockets) retained+=entry.second->buffer.capacity()+entry.second->pendingWrite.capacity()+entry.second->pendingRead.capacity()+(entry.second->ws ? entry.second->ws->storage() : 0);
if(retained+2*bytes>valueLimit) failLimit(s,"socket buffer storage limit (1 MiB)");
};
auto store=[&](const std::shared_ptr<Socket>& p,std::size_t output) {
auto id=std::string(p->datagram ? "udp:" : "tcp:")+std::to_string(nextSocket++);
sockets.emplace(id,p); put(output,id);
};
auto tlsResources=[&](const std::shared_ptr<Socket>& p) {
p->tls->releaseResources=[this,owned=p.get()]() {
if(owned->fd>=0) ::close(owned->fd);
owned->fd=-1;
for(auto& entry:sockets) { if(entry.second->fd>=0) ::close(entry.second->fd); entry.second->fd=-1; }
sockets.clear();
};
};
auto direction=[](short want) { return want==POLLIN ? "r" : want==POLLOUT ? "w" : ""; };
if(udp) {
#include "socket_udp.inc"
}
if(sub=="tlslisten") {
reserve(128,false,true);
if(!options.allowListen) fail(s,"socket listen requires --allow-listen and --allow-network");
if(options.tlsCertFile.empty() || options.tlsKeyFile.empty()) fail(s,"TLS listener requires --tls-cert-file and --tls-key-file");
auto host=val(3); if(host.empty() || host=="localhost") host="127.0.0.1";
socketHost(host,true,s); auto port=static_cast<int>(integer(val(4),0,65535,s));
int backlog=static_cast<int>(integer(val(5),1,128,s)); room();
auto address=socketNumeric(host,port); auto p=std::make_shared<Socket>(); creating=p; p->listener=true;
p->tls=std::make_unique<SocketTLS>(); tlsResources(p);
p->tls->serverContext(options,s); // Pair validation BEFORE socket/bind.
p->fd=::socket(address.address.ss_family,SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC,0);
if(p->fd<0) osError("create");
if(address.address.ss_family==AF_INET6) { int yes=1; if(setsockopt(p->fd,IPPROTO_IPV6,IPV6_V6ONLY,&yes,sizeof(yes))<0) osError("IPv6-only"); }
if(::bind(p->fd,reinterpret_cast<const sockaddr*>(&address.address),address.length)<0) osError("bind");
if(::listen(p->fd,backlog)<0) osError("listen");
store(p,2); return true;
}
if(sub=="wait") {
reserve(32768);
// Copy small validated interests before replacing a possible aliased output.
const auto& input=jsonValue(frame,a[3],s);
if(!input.is_array() || input.empty() || input.size()>32) fail(s,"socket wait requires 1..32 interests");
std::vector<std::shared_ptr<Socket>> owners; std::vector<std::string> keys,events;
std::vector<pollfd> polls; std::set<std::string> seen;
for(const auto& item:input) {
if(!item.is_object() || item.size()!=2 || !item.contains("handle") || !item.contains("events") ||
!item["handle"].is_string() || !item["events"].is_string()) fail(s,"invalid socket wait interest");
auto key=item["handle"].get<std::string>(), event=item["events"].get<std::string>();
if(event!="r" && event!="w" && event!="rw") fail(s,"invalid socket wait events");
if(!seen.insert(key).second) fail(s,"duplicate socket wait handle");
auto p=get(key);
if((p->listener || p->handshaking || p->ws) && event!="r") fail(s,"listener/handshake wait requires r");
if(!p->ws && p->tls && !p->pendingRead.empty() && p->readWant==POLLOUT && event!="r") fail(s,"pending TLS read/WANT_WRITE requires r interest");
if(!p->ws && p->tls && !p->pendingWrite.empty() && event!="w") fail(s,"pending TLS write requires w interest");
short requested=static_cast<short>((event.find('r')!=std::string::npos ? POLLIN : 0)|(event.find('w')!=std::string::npos ? POLLOUT : 0));
if(p->tls && p->want && (p->handshaking || !p->pendingWrite.empty() || (!p->pendingRead.empty() && p->readWant==POLLOUT))) requested=p->want;
if(p->ws) {
if(p->tls && !p->pendingRead.empty() && p->readWant==POLLOUT) requested=POLLOUT;
else if(!p->pendingWrite.empty()) requested=p->want ? p->want : POLLOUT;
else requested=p->ws->queue.empty() ? POLLIN : POLLOUT;
}
owners.push_back(p); keys.push_back(key); events.push_back(event); polls.push_back({p->fd,requested,0});
}
auto ms=static_cast<std::size_t>(integer(val(4),0,30000,s)); timeRoom(ms); socketTime+=ms;
auto deadline=SocketClock::now()+std::chrono::milliseconds(ms);
for(const auto& p:owners) {
if(p->handshaking) deadline=std::min(deadline,p->handshakeDeadline);
if(p->ws && p->ws->timed()) deadline=std::min(deadline,p->ws->deadline);
}
for(;;) {
bool immediate=false;
for(std::size_t i=0;i<owners.size();++i) {
auto& p=owners[i];
if(((!p->ws || (p->pendingWrite.empty() && p->ws->queue.empty())) && events[i].find('r')!=std::string::npos && (!p->buffer.empty() || p->eof || (p->tls && p->tls->ssl && p->readWant!=POLLOUT && SSL_pending(p->tls->ssl)>0))) ||
(p->handshaking && SocketClock::now()>=p->handshakeDeadline) ||
(p->ws && (p->wsBuffered() || (p->ws->timed() && SocketClock::now()>=p->ws->deadline)))) immediate=true;
}
auto ns=std::chrono::duration_cast<std::chrono::nanoseconds>(deadline-SocketClock::now()).count();
int delay=immediate ? 0 : static_cast<int>(std::max<std::int64_t>(0,(ns+999999)/1000000));
int rc=::poll(polls.data(),polls.size(),delay);
if(rc<0 && errno==EINTR) { if(SocketClock::now()<deadline) continue; rc=0; }
if(rc<0) osError("poll");
Json out=Json::array();
for(std::size_t i=0;i<owners.size();++i) {
auto& p=owners[i]; auto re=polls[i].revents;
bool expired=(p->handshaking && SocketClock::now()>=p->handshakeDeadline) || (p->ws && p->ws->timed() && SocketClock::now()>=p->ws->deadline);
bool error=expired || (re&(POLLERR|POLLNVAL));
bool ready=(re&(polls[i].events|POLLERR|POLLHUP|POLLNVAL))!=0;
bool read=events[i].find('r')!=std::string::npos && ((p->tls && (p->handshaking || (!p->pendingRead.empty() && p->readWant==POLLOUT)) ? ready : (re&(POLLIN|POLLHUP|POLLERR))) || !p->buffer.empty() || p->eof || (p->tls && p->tls->ssl && p->readWant!=POLLOUT && SSL_pending(p->tls->ssl)>0));
if(p->ws) read=ready || p->wsBuffered() || (p->tls && p->tls->ssl && p->pendingWrite.empty() && p->ws->queue.empty() && p->readWant!=POLLOUT && SSL_pending(p->tls->ssl)>0);
bool write=events[i].find('w')!=std::string::npos && (p->tls && p->want && !p->pendingWrite.empty() ? ready : (re&(POLLOUT|POLLHUP|POLLERR))!=0);
if(expired) { if(p->fd>=0) ::close(p->fd); p->fd=-1; sockets.erase(keys[i]); }
if(read || write || error) out.push_back({{"handle",keys[i]},{"read",read},{"write",write},{"eof",p->eof && p->buffer.empty()},{"error",error},{"handshake",p->handshaking},{"expired",expired}});
}
if(!out.empty() || SocketClock::now()>=deadline || ms==0) { result(std::move(out)); return true; }
}
}
if(sub=="tryaccept") {
reserve(256,true);
auto p=get(val(4)); if(p->datagram || !p->listener) fail(s,"socket tryaccept requires TCP listener");
auto ms=static_cast<std::size_t>(integer(val(5),1,30000,s)); room(); if(p->tls) timeRoom(ms);
auto client=std::make_shared<Socket>(); creating=client;
client->fd=::accept4(p->fd,nullptr,nullptr,SOCK_NONBLOCK|SOCK_CLOEXEC);
if(client->fd<0) {
if(!socketWouldBlock()) osError("accept");
put(3,""); result({{"status","wouldblock"}}); return true;
}
if(p->tls) {
socketTime+=ms; client->handshaking=true; client->want=POLLIN;
client->handshakeDeadline=SocketClock::now()+std::chrono::milliseconds(ms);
client->tls=std::make_unique<SocketTLS>(); tlsResources(client);
client->tls->serverConnection(client->fd,p->tls->context,s);
}
store(client,3); result({{"status",client->handshaking ? "handshake" : "progress"}}); return true;
}
bool reading=sub=="tryrecv";
std::size_t max=reading ? static_cast<std::size_t>(integer(val(5),1,65536,s)) : 0;
reserve(reading ? max : 256,reading);
auto key=val(reading ? 4 : 3); auto p=get(key);
if(p->datagram || p->listener) fail(s,"socket operation requires TCP connection");
if(p->ws) fail(s,"raw socket I/O forbidden on websocket");
if(p->handshaking && SocketClock::now()>=p->handshakeDeadline) { sockets.erase(key); fail(s,"socket TLS handshake timeout"); }
if(sub=="handshake") {
if(!p->tls) fail(s,"socket handshake requires TLS connection");
int n=1;
if(p->handshaking) {
try { n=p->tls->step(3,nullptr,0,p->want,s); }
catch(...) { sockets.erase(key); throw; }
if(SocketClock::now()>=p->handshakeDeadline) { sockets.erase(key); fail(s,"socket TLS handshake timeout"); }
if(n>0) p->handshaking=false;
}
result({{"status",n>0 ? "progress" : "wouldblock"},{"want",direction(p->want)}}); return true;
}
if(p->handshaking) fail(s,"socket TLS handshake incomplete");
if(sub=="trysend") {
auto data=val(4); if(data.size()>65536) fail(s,"socket trysend size limit (65536)");
// A read awaiting input may coexist with a write. Keep its owned retry
// arguments intact. Read/WANT_WRITE is conservatively exclusive.
if(!p->pendingRead.empty() && p->readWant!=POLLIN) fail(s,"socket pending TLS read/WANT_WRITE conflicts with write");
if(!p->pendingWrite.empty() && !data.empty()) fail(s,"socket pending write requires empty resume data");
std::size_t accepted=0;
if(!p->pendingWrite.empty()) storage(p->pendingWrite.size());
if(p->pendingWrite.empty()) {
bytesRoom(data.size()); storage(data.size());
p->pendingWrite=std::string(data); accepted=data.size(); socketBytes+=accepted;
}
ssize_t n=0;
if(!p->pendingWrite.empty()) {
try {
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 && !p->tls) { if(!socketWouldBlock()) osError("send"); p->want=POLLOUT; }
} catch(...) { sockets.erase(key); throw; }
if(n>0) {
std::string remaining(p->pendingWrite.data()+n,p->pendingWrite.size()-static_cast<std::size_t>(n));
p->pendingWrite.swap(remaining); p->want=p->pendingWrite.empty() ? p->readWant : 0;
}
}
result({{"status",n<0 ? "wouldblock" : "progress"},{"accepted",accepted},{"count",std::max<ssize_t>(0,n)},{"pending",!p->pendingWrite.empty()},{"want",direction(p->pendingWrite.empty() ? 0 : p->want)}}); return true;
}
if(p->tls && !p->pendingWrite.empty()) fail(s,"socket pending TLS write conflicts with read");
if(!p->pendingRead.empty() && p->readMax!=max) fail(s,"socket pending TLS read requires same maxbytes");
std::string data; ssize_t n=-1;
if(!p->buffer.empty()) {
auto size=std::min(max,p->buffer.size()); data=p->buffer.substr(0,size);
p->buffer=std::string(p->buffer.data()+size,p->buffer.size()-size); n=static_cast<ssize_t>(size);
} else if(p->eof) n=0;
else {
auto size=std::min<std::size_t>(8192,max); bytesRoom(size); storage(size);
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; if(!attempt) p->want=POLLIN;
}
if(attempt) {
if(p->pendingRead.empty()) { p->pendingRead=std::string(size,'\0'); p->readMax=max; }
try {
n=p->tls ? p->tls->step(1,p->pendingRead.data(),static_cast<int>(size),p->want,s) : ::recv(p->fd,p->pendingRead.data(),size,0);
if(p->tls) p->readWant=p->want;
if(n<0 && !p->tls) { if(!socketWouldBlock()) osError("recv"); p->want=POLLIN; }
} catch(...) {
// A failed transport read invalidates every copied handle,
// plain or TLS. Preflight checks remain outside this catch.
sockets.erase(key); throw;
}
if(n>0) { socketBytes+=static_cast<std::size_t>(n); data.assign(p->pendingRead.data(),static_cast<std::size_t>(n)); }
if(n>=0 || !p->tls) { std::string().swap(p->pendingRead); if(n>=0) p->want=0; }
if(n==0) p->eof=true;
}
}
put(3,data); result({{"status",n<0 ? "wouldblock" : n==0 ? "eof" : "progress"},{"count",data.size()},{"eof",p->eof && p->buffer.empty()},{"want",direction(p->want)}}); return true;
}