restored/sockets.inc
← All source files · Download this file
13,864 bytes · SHA256 3d64b748f868c125c205a019103f1c013a51b21ae93c8ee20594d60a69de54c0
// Linux TCP and verified TLS clients; isolated from interpreter dispatch.
static bool foundationOperation(const std::string&);
static void foundationValidate(const CodeLine&);
static void socketValidate(const CodeLine& c) {
count(c,2,7); const auto& sub=c.arg[1].text;
if(foundationOperation(sub)) { foundationValidate(c); return; }
if(sub=="connect" || sub=="tlsconnect" || sub=="listen") { count(c,6,6); destination(c.arg[2],c.source); }
else if(sub=="accept") { count(c,5,5); destination(c.arg[2],c.source); }
else if(sub=="sendall") count(c,5,5);
else if(sub=="recv" || sub=="readline") { count(c,7,7); destination(c.arg[2],c.source); destination(c.arg[3],c.source); }
else if(sub=="endpoint") { count(c,5,5); destination(c.arg[2],c.source); destination(c.arg[3],c.source); }
else if(sub=="close") count(c,3,3);
else fail(c.source,"unknown socket operation");
}
using SocketClock=std::chrono::steady_clock;
static void socketDeadline(SocketClock::time_point deadline,const Source& s) {
if(SocketClock::now()>=deadline) fail(s,"socket timeout");
}
static void socketWait(int fd,short events,SocketClock::time_point deadline,const Source& s) {
for(;;) {
socketDeadline(deadline,s);
auto ns=std::chrono::duration_cast<std::chrono::nanoseconds>(deadline-SocketClock::now()).count();
int ms=static_cast<int>(std::max<std::int64_t>(1,(ns+999999)/1000000));
pollfd p{fd,events,0}; int rc=::poll(&p,1,ms);
if(rc<0 && errno==EINTR) continue;
if(rc<0) {
if(errno==ENOMEM || errno==ENOBUFS) failLimit(s,"socket poll resource limit");
fail(s,"socket poll failed: "+std::string(std::strerror(errno)));
}
if(rc>0) { if(p.revents&POLLNVAL) fail(s,"socket invalid descriptor"); return; }
}
}
#include "socket_dns.inc"
#include "socket_tls.inc"
#include "websocket_protocol.inc"
struct vScript::Socket {
int fd=-1;
bool listener=false, eof=false;
bool datagram=false, handshaking=false;
short want=0, readWant=0;
std::string pendingWrite, pendingRead;
std::size_t readMax=0;
SocketClock::time_point handshakeDeadline{};
std::string buffer;
std::unique_ptr<SocketTLS> tls;
std::unique_ptr<SocketWebSocket> ws;
bool wsBuffered() const { return ws && ws->parseReady && pendingWrite.empty() && ws->queue.empty() && !(tls && !pendingRead.empty() && readWant==POLLOUT); }
~Socket() { tls.reset(); if(fd>=0) ::close(fd); }
};
#include "socket_foundations.inc"
#include "websocket.inc"
bool vScript::socketCommand(Frame& frame,const CodeLine& c) {
const auto& a=c.arg; const auto& s=c.source; const auto& sub=a[1].text;
if(foundationOperation(sub)) return foundationCommand(frame,c);
auto socketOS=[&](const Source& source,const std::string& operation) {
int error=errno;
if(error==ENOMEM || error==ENOBUFS || error==EMFILE || error==ENFILE) {
// Fatal: release owned descriptors before diagnostic machinery runs.
// In particular sanitizer diagnostics themselves need spare FDs.
sockets.clear();
failLimit(source,"socket OS resource limit: "+operation);
}
fail(source,"socket "+operation+": "+std::strerror(error));
};
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 dest=[&](std::size_t i) { auto& v=lookup(frame,a[i],s); if(v.array || v.structured) fail(s,"socket destination must be scalar"); };
auto outputs=[&](std::size_t max) {
dest(2);
if(sub=="recv" || sub=="readline" || sub=="endpoint") { dest(3); if(a[2].text==a[3].text) fail(s,"socket destinations must differ"); }
// Conservative reservation BEFORE effects, including all retained capacities.
std::size_t total=frame.extraBytes+max+256;
for(const auto& entry:frame.vars) total+=variableSize(entry.second)+entry.first.capacity();
if(total>frameLimit) failLimit(s,"socket frame reservation limit");
};
if(sub=="connect" || sub=="tlsconnect" || sub=="listen" || sub=="accept" || sub=="endpoint") outputs(128);
if(!options.allowNetwork) fail(s,"socket operations require --allow-network (not a sandbox)");
if(sub=="listen" && !options.allowListen) fail(s,"socket listen requires --allow-listen and --allow-network");
if(++socketCalls>4096) failLimit(s,"socket call limit (4096)");
auto handle=[&](std::size_t i) {
auto key=val(i); auto it=sockets.find(key);
if(it==sockets.end()) fail(s,"invalid or closed socket handle");
if(it->second->datagram) fail(s,"socket operation requires TCP handle");
if(it->second->ws && sub!="close" && sub!="endpoint") fail(s,"raw socket I/O forbidden on websocket");
return it;
};
auto timeout=[&](std::size_t i) {
auto ms=static_cast<std::size_t>(integer(val(i),1,30000,s));
if(ms>60000-socketTime) failLimit(s,"socket cumulative timeout reservation limit (60000 ms)");
socketTime+=ms;
return SocketClock::now()+std::chrono::milliseconds(ms);
};
auto room=[&]() { if(sockets.size()>=32) failLimit(s,"socket handle limit (32)"); };
auto store=[&](const std::shared_ptr<Socket>& p) {
auto id="tcp:"+std::to_string(nextSocket++); sockets.emplace(id,p); put(2,id);
};
if(sub=="connect" || sub=="tlsconnect" || sub=="listen") {
auto originalHost=val(3), host=originalHost;
bool listener=sub=="listen";
if(host=="localhost" || (host.empty() && listener)) host="127.0.0.1";
auto port=static_cast<int>(integer(val(4),listener ? 0 : 1,65535,s));
socketHost(host,listener,s);
int backlog=0; SocketClock::time_point deadline;
if(listener) backlog=static_cast<int>(integer(val(5),1,128,s)); else deadline=timeout(5);
room();
std::vector<SocketAddress> addresses;
if(listener) addresses.push_back(socketNumeric(host,port));
else {
if(!socketNumeric(host,port).length && ++dnsCalls>128) failLimit(s,"socket DNS lookup limit (128)");
try { addresses=socketResolve(host,port,options,deadline,s,[&]() { sockets.clear(); }); }
catch(const Error& e) { if(e.fatal) sockets.clear(); throw; }
}
std::shared_ptr<Socket> p;
int lastError=ECONNREFUSED;
for(const auto& address:addresses) {
if(!listener) socketDeadline(deadline,s);
p=std::make_shared<Socket>(); p->listener=listener;
int family=address.address.ss_family;
p->fd=::socket(family,SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC,0);
if(p->fd<0) socketOS(s,"create");
if(listener) {
if(family==AF_INET6) { int yes=1; if(setsockopt(p->fd,IPPROTO_IPV6,IPV6_V6ONLY,&yes,sizeof(yes))<0) socketOS(s,"IPv6-only"); }
if(::bind(p->fd,reinterpret_cast<const sockaddr*>(&address.address),address.length)<0) socketOS(s,"bind");
if(::listen(p->fd,backlog)<0) socketOS(s,"listen");
break;
}
int rc=::connect(p->fd,reinterpret_cast<const sockaddr*>(&address.address),address.length);
int error=rc==0 ? 0 : errno;
if(rc<0 && (error==EINPROGRESS || error==EINTR || error==EALREADY)) {
socketWait(p->fd,POLLOUT,deadline,s);
socklen_t len=sizeof(error);
if(getsockopt(p->fd,SOL_SOCKET,SO_ERROR,&error,&len)<0) socketOS(s,"connect status");
}
if(!error) break;
if(error==ENOMEM || error==ENOBUFS || error==EMFILE || error==ENFILE) { errno=error; socketOS(s,"connect"); }
lastError=error; p.reset();
}
if(!p) { errno=lastError; socketOS(s,"connect"); }
if(!listener) socketDeadline(deadline,s);
if(sub=="tlsconnect") {
p->tls=std::make_unique<SocketTLS>();
p->tls->releaseResources=[this,owned=p.get()]() {
if(owned->fd>=0) { ::close(owned->fd); owned->fd=-1; }
sockets.clear();
};
try { p->tls->connect(p->fd,originalHost,options,deadline,s); }
catch(const Error& e) { if(e.fatal) sockets.clear(); throw; }
}
store(p); return true;
}
if(sub=="close") { auto it=handle(2); if(it->second->tls && !it->second->handshaking && it->second->pendingWrite.empty() && it->second->pendingRead.empty()) it->second->tls->close(); sockets.erase(it); return true; }
if(sub=="endpoint") {
auto p=handle(4)->second; sockaddr_storage addr{}; socklen_t len=sizeof(addr);
if(getsockname(p->fd,reinterpret_cast<sockaddr*>(&addr),&len)<0) socketOS(s,"endpoint");
char text[INET6_ADDRSTRLEN]; unsigned port;
if(addr.ss_family==AF_INET) { auto* x=reinterpret_cast<sockaddr_in*>(&addr); inet_ntop(AF_INET,&x->sin_addr,text,sizeof(text)); port=ntohs(x->sin_port); }
else { auto* x=reinterpret_cast<sockaddr_in6*>(&addr); inet_ntop(AF_INET6,&x->sin6_addr,text,sizeof(text)); port=ntohs(x->sin6_port); }
put(2,text); put(3,std::to_string(port)); return true;
}
if(sub=="accept") {
auto p=handle(3)->second; if(!p->listener) fail(s,"socket accept requires listener");
if(p->tls) fail(s,"TLS listener requires tryaccept and handshake");
auto deadline=timeout(4); room(); auto client=std::make_shared<Socket>();
for(;;) {
socketDeadline(deadline,s);
client->fd=::accept4(p->fd,nullptr,nullptr,SOCK_NONBLOCK|SOCK_CLOEXEC);
if(client->fd>=0) break;
if(errno==EINTR) continue;
if(errno==EAGAIN || errno==EWOULDBLOCK) { socketWait(p->fd,POLLIN,deadline,s); continue; }
socketOS(s,"accept");
}
store(client); return true;
}
bool reading=sub=="recv" || sub=="readline";
auto connection=handle(reading ? 4 : 2);
auto p=connection->second;
auto connectionKey=connection->first;
if(p->listener) fail(s,"socket operation requires connection, not listener");
if(p->handshaking) {
if(SocketClock::now()>=p->handshakeDeadline) { sockets.erase(connectionKey); fail(s,"socket TLS handshake timeout"); }
fail(s,"socket TLS handshake incomplete");
}
if(!p->pendingWrite.empty() || !p->pendingRead.empty()) fail(s,"socket pending operation conflicts with legacy I/O");
if(!reading) {
auto data=val(3); auto deadline=timeout(4);
if(data.size()>16*valueLimit-socketBytes) failLimit(s,"socket cumulative byte limit (16 MiB)");
// Charge attempted writes; retries cannot evade the total bound.
socketBytes+=data.size(); std::size_t sent=0;
try {
while(sent<data.size()) {
socketDeadline(deadline,s);
auto n=p->tls ? p->tls->perform(2,data.data()+sent,static_cast<int>(data.size()-sent),deadline,s) :
::send(p->fd,data.data()+sent,data.size()-sent,MSG_NOSIGNAL);
if(n>0) { sent+=static_cast<std::size_t>(n); if(p->tls) socketDeadline(deadline,s); continue; }
if(n<0 && errno==EINTR) continue;
if(n<0 && (errno==EAGAIN || errno==EWOULDBLOCK)) { socketWait(p->fd,POLLOUT,deadline,s); continue; }
socketOS(s,"send");
}
} catch(const Error& e) { if(p->tls) sockets.erase(connectionKey); if(e.fatal) sockets.clear(); throw Error(s,std::string(e.what())+"; sent="+std::to_string(sent)+" of "+std::to_string(data.size())+" bytes (do not blindly retry)",e.fatal); }
return true;
}
auto max=static_cast<std::size_t>(integer(val(5),1,65536,s)); outputs(max);
auto deadline=timeout(6);
for(;;) {
auto newline=p->buffer.find('\n');
bool line=sub=="readline";
if((!line && !p->buffer.empty()) || (line && newline!=std::string::npos && newline<max) || p->eof) {
auto n=std::min(max,p->buffer.size()); bool terminated=line && newline<n;
if(terminated) n=newline+1;
put(2,p->buffer.substr(0,terminated ? n-1 : n));
put(3,p->eof && n==p->buffer.size() ? "1" : "0");
p->buffer=std::string(p->buffer.data()+n,p->buffer.size()-n);
return true;
}
if(line && p->buffer.size()>=max) fail(s,"socket line limit (buffer preserved; drain with recv)");
std::size_t want=std::min<std::size_t>(8192,max-p->buffer.size());
if(want>16*valueLimit-socketBytes) failLimit(s,"socket cumulative byte limit (16 MiB)");
// Rebuild each append to bound retained capacity, not just logical length.
std::size_t retained=0; for(const auto& entry:sockets) retained+=entry.second->buffer.capacity()+(entry.second->pendingRead.empty() ? 0 : entry.second->pendingRead.capacity())+(entry.second->pendingWrite.empty() ? 0 : entry.second->pendingWrite.capacity())+(entry.second->ws ? entry.second->ws->storage() : 0);
if(retained+2*(p->buffer.size()+want)>valueLimit) failLimit(s,"socket buffer storage limit (1 MiB)");
char bytes[8192]; ssize_t n;
if(p->tls) {
try { n=p->tls->perform(1,bytes,static_cast<int>(want),deadline,s); }
catch(const Error& e) { sockets.erase(connectionKey); if(e.fatal) sockets.clear(); throw; }
} else { socketDeadline(deadline,s); n=::recv(p->fd,bytes,want,0); }
if(n>0) { socketBytes+=static_cast<std::size_t>(n); std::string fresh=p->buffer; fresh.append(bytes,static_cast<std::size_t>(n)); p->buffer.swap(fresh); continue; }
if(n==0) { p->eof=true; continue; }
if(errno==EINTR) continue;
if(errno==EAGAIN || errno==EWOULDBLOCK) { socketWait(p->fd,POLLIN,deadline,s); continue; }
socketOS(s,"recv");
}
}