vieter-v
/
v
2
0
Fork 0
Code Issues Pull Requests Releases Activity

websocket: make compile

pull/5048/head^2
Enzo Baldisserri 2020-05-26 12:50:37 +02:00 committed by GitHub
parent e79adc0ba1
commit 145b125155
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 284 additions and 268 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cmd/tools/vtest-fixed.v
View File

@ -11,11 +11,14 @@ const (
skip_on_musl = [ skip_on_musl = [
'vlib/net/http/http_test.v', 'vlib/net/http/http_test.v',
'vlib/net/http/cookie_test.v', 'vlib/net/http/cookie_test.v',
'vlib/net/websocket/ws_test.v'
'vlib/sqlite/sqlite_test.v', 'vlib/sqlite/sqlite_test.v',
'vlib/clipboard/clipboard_test.v', 'vlib/clipboard/clipboard_test.v',
] ]
skip_on_linux = []string{} skip_on_linux = []string{}
skip_on_non_linux = []string{} skip_on_non_linux = [
'vlib/net/websocket/ws_test.v'
]
skip_on_windows = []string{} skip_on_windows = []string{}
skip_on_non_windows = []string{} skip_on_non_windows = []string{}
skip_on_macos = []string{} skip_on_macos = []string{}

46
vlib/net/websocket/handshake.v
View File

@ -1,18 +1,18 @@
module websocket module websocket
fn (mut ws Client) read_handshake(seckey string) { fn (mut ws Client) read_handshake(seckey string) {
l.d("reading handshake...") l.d('reading handshake...')
mut bytes_read := 0 mut bytes_read := 0
max_buffer := 1024 max_buffer := 1024
buffer_size := 1 buffer_size := 1
mut buffer := malloc(max_buffer) mut buffer := malloc(max_buffer)
for bytes_read <= max_buffer { for bytes_read <= max_buffer {
res := ws.read_from_server(buffer + bytes_read, buffer_size) res := ws.read_from_server(buffer + bytes_read, buffer_size)
if res == 0 || res == -1 { if res == 0 || res == -1 {
l.f("read_handshake: Failed to read handshake.") l.f('read_handshake: Failed to read handshake.')
} }
if buffer[bytes_read] == `\n` && buffer[bytes_read-1] == `\r` && buffer[bytes_read-2] == `\n` && buffer[bytes_read-3] == `\r` { if buffer[bytes_read] == `\n` && buffer[bytes_read - 1] == `\r` && buffer[bytes_read -
2] == `\n` && buffer[bytes_read - 3] == `\r` {
break break
} }
bytes_read += buffer_size bytes_read += buffer_size
@ -22,51 +22,49 @@ fn (mut ws Client) read_handshake(seckey string){
} }
fn (mut ws Client) handshake_handler(handshake_response, seckey string) { fn (mut ws Client) handshake_handler(handshake_response, seckey string) {
l.d("handshake_handler:\r\n${handshake_response}") l.d('handshake_handler:\r\n${handshake_response}')
lines := handshake_response.split_into_lines() lines := handshake_response.split_into_lines()
header := lines[0] header := lines[0]
if !header.starts_with("HTTP/1.1 101") && !header.starts_with("HTTP/1.0 101") { if !header.starts_with('HTTP/1.1 101') && !header.starts_with('HTTP/1.0 101') {
l.f("handshake_handler: invalid HTTP status response code") l.f('handshake_handler: invalid HTTP status response code')
} }
for i in 1 .. lines.len { for i in 1 .. lines.len {
if lines[i].len <= 0 || lines[i] == "\r\n" { if lines[i].len <= 0 || lines[i] == '\r\n' {
continue continue
} }
keys := lines[i].split(":") keys := lines[i].split(':')
match keys[0] { match keys[0] {
"Upgrade", "upgrade" { 'Upgrade', 'upgrade' {
ws.flags << .has_upgrade ws.flags << .has_upgrade
} }
"Connection", "connection" { 'Connection', 'connection' {
ws.flags << .has_connection ws.flags << .has_connection
} }
"Sec-WebSocket-Accept", "sec-websocket-accept" { 'Sec-WebSocket-Accept', 'sec-websocket-accept' {
l.d("comparing hashes") l.d('comparing hashes')
l.d("seckey: ${seckey}") l.d('seckey: ${seckey}')
challenge := create_key_challenge_response(seckey) challenge := create_key_challenge_response(seckey)
l.d("challenge: ${challenge}") l.d('challenge: ${challenge}')
l.d("response: ${keys[1]}") l.d('response: ${keys[1]}')
if keys[1].trim_space() != challenge { if keys[1].trim_space() != challenge {
l.e("handshake_handler: Sec-WebSocket-Accept header does not match computed sha1/base64 response.") l.e('handshake_handler: Sec-WebSocket-Accept header does not match computed sha1/base64 response.')
} }
ws.flags << .has_accept ws.flags << .has_accept
unsafe { unsafe {
challenge.free() challenge.free()
} }
} else {} }
else {}
} }
unsafe { unsafe {
keys.free() keys.free()
} }
} }
if ws.flags.len < 3 { if ws.flags.len < 3 {
ws.close(1002, "invalid websocket HTTP headers") ws.close(1002, 'invalid websocket HTTP headers')
l.e("invalid websocket HTTP headers") l.e('invalid websocket HTTP headers')
} }
l.i("handshake successful!") l.i('handshake successful!')
unsafe { unsafe {
handshake_response.free() handshake_response.free()
lines.free() lines.free()

44
vlib/net/websocket/ssl.v
View File

@ -1,45 +1,69 @@
module websocket module websocket
// On linux, prefer a localy build openssl, because it is
// much more likely for it to be newer, than the system
// openssl from libssl-dev. If there is no local openssl,
// the next flag is harmless, since it will still use the
// (older) system openssl.
#flag linux -I/usr/local/include/openssl -L/usr/local/lib
#flag -lssl #flag -lssl
// MacPorts
#flag darwin -I/opt/local/include
#flag darwin -L/opt/local/lib
// Brew
#flag darwin -I/usr/local/opt/openssl/include
#flag darwin -L/usr/local/opt/openssl/lib
#include <openssl/rand.h> #include <openssl/rand.h>
#include <openssl/ssl.h> #include <openssl/ssl.h>
#include <openssl/err.h> #include <openssl/err.h>
struct SSL_CTX {
}
struct SSL {
}
struct SSL_METHOD {
}
struct C.SSL_CTX
struct C.SSL
struct C.SSL_METHOD
fn C.SSL_load_error_strings() fn C.SSL_load_error_strings()
fn C.SSL_library_init() fn C.SSL_library_init()
fn C.SSLv23_client_method() &C.SSL_METHOD fn C.SSLv23_client_method() &C.SSL_METHOD
fn C.SSL_CTX_new() &C.SSL_CTX fn C.SSL_CTX_new() &C.SSL_CTX
fn C.SSL_new() &C.SSL fn C.SSL_new() &C.SSL
fn C.SSL_set_fd() int fn C.SSL_set_fd() int
fn C.SSL_connect() int fn C.SSL_connect() int
fn C.SSL_shutdown() fn C.SSL_shutdown()
fn C.SSL_free() fn C.SSL_free()
fn C.SSL_CTX_free() fn C.SSL_CTX_free()
fn C.SSL_write() int fn C.SSL_write() int
fn C.SSL_read() int fn C.SSL_read() int
fn (mut ws Client) connect_ssl() { fn (mut ws Client) connect_ssl() {
l.i("Using secure SSL connection") l.i('Using secure SSL connection')
C.SSL_load_error_strings() C.SSL_load_error_strings()
C.SSL_library_init() C.SSL_library_init()
ws.sslctx = C.SSL_CTX_new(C.SSLv23_client_method()) ws.sslctx = C.SSL_CTX_new(C.SSLv23_client_method())
if ws.sslctx == C.NULL { if ws.sslctx == 0 {
l.f("Couldn't get ssl context") l.f("Couldn't get ssl context")
} }
ws.ssl = C.SSL_new(ws.sslctx) ws.ssl = C.SSL_new(ws.sslctx)
if ws.ssl == C.NULL { if ws.ssl == 0 {
l.f("Couldn't create OpenSSL instance.") l.f("Couldn't create OpenSSL instance.")
} }
if C.SSL_set_fd(ws.ssl, ws.socket.sockfd) != 1 { if C.SSL_set_fd(ws.ssl, ws.socket.sockfd) != 1 {
l.f("Couldn't assign ssl to socket.") l.f("Couldn't assign ssl to socket.")
} }
if C.SSL_connect(ws.ssl) != 1 { if C.SSL_connect(ws.ssl) != 1 {
l.f("Couldn't connect using SSL.") l.f("Couldn't connect using SSL.")
} }

49
vlib/net/websocket/utf8.v
View File

@ -3,17 +3,21 @@ module websocket
pub fn utf8_validate_str(str string) bool { pub fn utf8_validate_str(str string) bool {
return utf8_validate(str.str, str.len) return utf8_validate(str.str, str.len)
} }
struct Utf8State { struct Utf8State {
mut: mut:
index int index int
subindex int subindex int
failed bool failed bool
} }
pub fn utf8_validate(data byteptr, len int) bool { pub fn utf8_validate(data byteptr, len int) bool {
mut state := Utf8State{} mut state := Utf8State{}
for i := 0; i < len; i++ { for i := 0; i < len; i++ {
s := data[i] s := data[i]
if s == 0 {break} if s == 0 {
break
}
state.next_state(s) state.next_state(s)
if state.failed { if state.failed {
return false return false
@ -23,15 +27,15 @@ pub fn utf8_validate(data byteptr, len int) bool {
return !state.failed && state.subindex <= 0 return !state.failed && state.subindex <= 0
} }
fn (mut s Utf8State) seq(r0 bool, r1 bool, is_tail bool) bool { fn (mut s Utf8State) seq(r0, r1, is_tail bool) bool {
if s.subindex == 0 || (s.index > 1 && s.subindex == 1) || (s.index >= 6 && s.subindex == 2) { if s.subindex == 0 || (s.index > 1 && s.subindex == 1) || (s.index >= 6 && s.subindex ==
2) {
if (s.subindex == 0 && r0) || (s.subindex == 1 && r1) || (s.subindex == 2 && is_tail) { if (s.subindex == 0 && r0) || (s.subindex == 1 && r1) || (s.subindex == 2 && is_tail) {
s.subindex++ s.subindex++
return true return true
} }
goto next goto next
} } else {
else {
s.failed = true s.failed = true
if is_tail { if is_tail {
s.index = 0 s.index = 0
@ -57,19 +61,32 @@ fn (mut s Utf8State) next_state (c byte) {
} }
is_tail := c >= 0x80 && c <= 0xBF is_tail := c >= 0x80 && c <= 0xBF
// sequence 2 // sequence 2
if s.index == 1 && s.seq(c >= 0xC2 && c <= 0xDF, false, is_tail) {return} if s.index == 1 && s.seq(c >= 0xC2 && c <= 0xDF, false, is_tail) {
return
}
// sequence 3 // sequence 3
if s.index == 2 && s.seq(c == 0xE0, c >= 0xA0 && c <= 0xBF, is_tail) {return} if s.index == 2 && s.seq(c == 0xE0, c >= 0xA0 && c <= 0xBF, is_tail) {
if s.index == 3 && s.seq(c >= 0xE1 && c <= 0xEC, c >= 0x80 && c <= 0xBF, is_tail) {return} return
if s.index == 4 && s.seq(c == 0xED, c >= 0x80 && c <= 0x9F, is_tail) {return} }
if s.index == 5 && s.seq(c >= 0xEE && c <= 0xEF, c >= 0x80 && c <= 0xBF, is_tail) {return} if s.index == 3 && s.seq(c >= 0xE1 && c <= 0xEC, c >= 0x80 && c <= 0xBF, is_tail) {
return
}
if s.index == 4 && s.seq(c == 0xED, c >= 0x80 && c <= 0x9F, is_tail) {
return
}
if s.index == 5 && s.seq(c >= 0xEE && c <= 0xEF, c >= 0x80 && c <= 0xBF, is_tail) {
return
}
// sequence 4 // sequence 4
if s.index == 6 && s.seq(c == 0xF0, c >= 0x90 && c <= 0xBF, is_tail) {return} if s.index == 6 && s.seq(c == 0xF0, c >= 0x90 && c <= 0xBF, is_tail) {
if s.index == 7 && s.seq(c >= 0xF1 && c <= 0xF3, c >= 0x80 && c <= 0xBF, is_tail) {return} return
if s.index == 8 && s.seq(c == 0xF4, c >= 0x80 && c <= 0x8F, is_tail) {return} }
if s.index == 7 && s.seq(c >= 0xF1 && c <= 0xF3, c >= 0x80 && c <= 0xBF, is_tail) {
return
}
if s.index == 8 && s.seq(c == 0xF4, c >= 0x80 && c <= 0x8F, is_tail) {
return
}
// we should never reach here // we should never reach here
s.failed = true s.failed = true
} }

7
vlib/net/websocket/utils.v
View File

@ -8,7 +8,6 @@ import encoding.base64
fn htonl64(payload_len u64) byteptr { fn htonl64(payload_len u64) byteptr {
mut ret := malloc(8) mut ret := malloc(8)
ret[0] = byte(((payload_len & (u64(0xff) << 56)) >> 56) & 0xff) ret[0] = byte(((payload_len & (u64(0xff) << 56)) >> 56) & 0xff)
ret[1] = byte(((payload_len & (u64(0xff) << 48)) >> 48) & 0xff) ret[1] = byte(((payload_len & (u64(0xff) << 48)) >> 48) & 0xff)
ret[2] = byte(((payload_len & (u64(0xff) << 40)) >> 40) & 0xff) ret[2] = byte(((payload_len & (u64(0xff) << 40)) >> 40) & 0xff)
@ -31,7 +30,7 @@ fn create_masking_key() []byte {
} }
fn create_key_challenge_response(seckey string) string { fn create_key_challenge_response(seckey string) string {
guid := "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" guid := '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
sha1buf := seckey + guid sha1buf := seckey + guid
hash := sha1.sum(sha1buf.bytes()) hash := sha1.sum(sha1buf.bytes())
hashstr := string(byteptr(hash.data)) hashstr := string(byteptr(hash.data))
@ -45,8 +44,8 @@ fn create_key_challenge_response(seckey string) string {
fn get_nonce() string { fn get_nonce() string {
mut nonce := []byte{} mut nonce := []byte{}
alphanum := "0123456789ABCDEFGHIJKLMNOPQRSTUVXYZabcdefghijklmnopqrstuvwxyz" alphanum := '0123456789ABCDEFGHIJKLMNOPQRSTUVXYZabcdefghijklmnopqrstuvwxyz'
for i in 0..18 { for _ in 0 .. 18 {
nonce << alphanum[rand.next(61)] nonce << alphanum[rand.next(61)]
} }
return string(byteptr(nonce.data)) return string(byteptr(nonce.data))

235
vlib/net/websocket/ws.v
View File

@ -8,15 +8,15 @@ import sync
import net.websocket.logger import net.websocket.logger
const ( const (
l = logger.new("ws") l = logger.new('ws')
) )
pub struct Client { pub struct Client {
retry int retry int
eb &eventbus.EventBus eb &eventbus.EventBus
is_ssl bool is_ssl bool
lock sync.Mutex = sync.new_mutex() lock &sync.Mutex = sync.new_mutex()
write_lock sync.Mutex = sync.new_mutex() write_lock &sync.Mutex = sync.new_mutex()
// subprotocol_len int // subprotocol_len int
// cwebsocket_subprotocol *subprotocol; // cwebsocket_subprotocol *subprotocol;
// cwebsocket_subprotocol *subprotocols[]; // cwebsocket_subprotocol *subprotocols[];
@ -93,20 +93,23 @@ pub fn new(uri string) &Client {
ws := &Client{ ws := &Client{
uri: uri uri: uri
state: .closed state: .closed
eb: eb, eb: eb
subscriber: eb.subscriber subscriber: eb.subscriber
is_ssl: uri.starts_with("wss") is_ssl: uri.starts_with('wss')
ssl: C.NULL ssl: 0
sslctx: C.NULL sslctx: 0
} }
return ws return ws
} }
fn C.sscanf() int fn C.sscanf() int
fn (ws &Client) parse_uri() &Uri { fn (ws &Client) parse_uri() &Uri {
u := urllib.parse(ws.uri) or {panic(err)} u := urllib.parse(ws.uri) or {
v := u.request_uri().split("?") panic(err)
querystring := if v.len > 1 {"?" + v[1]} else {""} }
v := u.request_uri().split('?')
querystring := if v.len > 1 { '?' + v[1] } else { '' }
return &Uri{ return &Uri{
hostname: u.hostname() hostname: u.hostname()
port: u.port() port: u.port()
@ -118,34 +121,29 @@ fn (ws &Client) parse_uri() &Uri {
pub fn (mut ws Client) connect() int { pub fn (mut ws Client) connect() int {
match ws.state { match ws.state {
.connected { .connected {
l.f("connect: websocket already connected") l.f('connect: websocket already connected')
} }
.connecting { .connecting {
l.f("connect: websocket already connecting") l.f('connect: websocket already connecting')
} }
.open { .open {
l.f("connect: websocket already open") l.f('connect: websocket already open')
} else { }
else {
// do nothing // do nothing
} }
} }
ws.lock.lock() ws.lock.lock()
ws.state = .connecting ws.state = .connecting
ws.lock.unlock() ws.lock.unlock()
uri := ws.parse_uri() uri := ws.parse_uri()
nonce := get_nonce() nonce := get_nonce()
seckey := base64.encode(nonce) seckey := base64.encode(nonce)
ai_family := C.AF_INET ai_family := C.AF_INET
ai_socktype := C.SOCK_STREAM ai_socktype := C.SOCK_STREAM
l.d('handshake header:')
l.d("handshake header:") handshake := 'GET ${uri.resource}${uri.querystring} HTTP/1.1\r\nHost: ${uri.hostname}:${uri.port}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: ${seckey}\r\nSec-WebSocket-Version: 13\r\n\r\n'
handshake := "GET ${uri.resource}${uri.querystring} HTTP/1.1\r\nHost: ${uri.hostname}:${uri.port}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: ${seckey}\r\nSec-WebSocket-Version: 13\r\n\r\n"
l.d(handshake) l.d(handshake)
socket := net.new_socket(ai_family, ai_socktype, 0) or { socket := net.new_socket(ai_family, ai_socktype, 0) or {
l.f(err) l.f(err)
return -1 return -1
@ -155,34 +153,26 @@ pub fn (mut ws Client) connect() int {
l.f(err) l.f(err)
return -1 return -1
} }
optval := 1 optval := 1
ws.socket.setsockopt(C.SOL_SOCKET, C.SO_KEEPALIVE, &optval) or { ws.socket.setsockopt(C.SOL_SOCKET, C.SO_KEEPALIVE, &optval) or {
l.f(err) l.f(err)
return -1 return -1
} }
if ws.is_ssl { if ws.is_ssl {
ws.connect_ssl() ws.connect_ssl()
} }
ws.lock.lock() ws.lock.lock()
ws.state = .connected ws.state = .connected
ws.lock.unlock() ws.lock.unlock()
res := ws.write_to_server(handshake.str, handshake.len) res := ws.write_to_server(handshake.str, handshake.len)
if res <= 0 { if res <= 0 {
l.f("Handshake failed.") l.f('Handshake failed.')
} }
ws.read_handshake(seckey) ws.read_handshake(seckey)
ws.lock.lock() ws.lock.lock()
ws.state = .open ws.state = .open
ws.lock.unlock() ws.lock.unlock()
ws.send_open_event() ws.send_open_event()
unsafe { unsafe {
handshake.free() handshake.free()
nonce.free() nonce.free()
@ -193,36 +183,33 @@ pub fn (mut ws Client) connect() int {
pub fn (mut ws Client) close(code int, message string) { pub fn (mut ws Client) close(code int, message string) {
if ws.state != .closed && ws.socket.sockfd > 1 { if ws.state != .closed && ws.socket.sockfd > 1 {
ws.lock.lock() ws.lock.lock()
ws.state = .closing ws.state = .closing
ws.lock.unlock() ws.lock.unlock()
mut code32 := 0 mut code32 := 0
if code > 0 { if code > 0 {
_code := C.htons(code) code_ := C.htons(code)
message_len := message.len + 2 message_len := message.len + 2
mut close_frame := [`0`].repeat(message_len) mut close_frame := [`0`].repeat(message_len)
close_frame[0] = _code & 0xFF close_frame[0] = code_ & 0xFF
close_frame[1] = (_code >> 8) close_frame[1] = (code_ >> 8)
code32 = (close_frame[0] << 8) + close_frame[1] code32 = (close_frame[0] << 8) + close_frame[1]
for i in 0 .. message.len { for i in 0 .. message.len {
close_frame[i + 2] = message[i] close_frame[i + 2] = message[i]
} }
ws.send_control_frame(.close, "CLOSE", close_frame) ws.send_control_frame(.close, 'CLOSE', close_frame)
} else { } else {
ws.send_control_frame(.close, "CLOSE", []) ws.send_control_frame(.close, 'CLOSE', [])
} }
if ws.ssl != 0 {
if ws.ssl != C.NULL {
C.SSL_shutdown(ws.ssl) C.SSL_shutdown(ws.ssl)
C.SSL_free(ws.ssl) C.SSL_free(ws.ssl)
if ws.sslctx != C.NULL { if ws.sslctx != 0 {
C.SSL_CTX_free(ws.sslctx) C.SSL_CTX_free(ws.sslctx)
} }
} else { } else {
if C.shutdown(ws.socket.sockfd, C.SHUT_WR) == -1 { if C.shutdown(ws.socket.sockfd, C.SHUT_WR) == -1 {
l.e("Unabled to shutdown websocket.") l.e('Unabled to shutdown websocket.')
} }
mut buf := [`0`] mut buf := [`0`]
for ws.read_from_server(buf.data, 1) > 0 { for ws.read_from_server(buf.data, 1) > 0 {
@ -232,34 +219,35 @@ pub fn (mut ws Client) close(code int, message string){
buf.free() buf.free()
} }
if C.close(ws.socket.sockfd) == -1 { if C.close(ws.socket.sockfd) == -1 {
//ws.send_close_event()(websocket, 1011, strerror(errno)); // ws.send_close_event()(websocket, 1011, strerror(C.errno));
} }
} }
ws.fragments = [] ws.fragments = []
ws.send_close_event() ws.send_close_event()
ws.lock.lock() ws.lock.lock()
ws.state = .closed ws.state = .closed
ws.lock.unlock() ws.lock.unlock()
unsafe { unsafe {
} }
// TODO impl autoreconnect // TODO impl autoreconnect
} }
} }
pub fn (mut ws Client) write(payload byteptr, payload_len int, code OPCode) int { pub fn (mut ws Client) write(payload byteptr, payload_len int, code OPCode) int {
mut bytes_written := -1
if ws.state != .open { if ws.state != .open {
ws.send_error_event("WebSocket closed. Cannot write.") ws.send_error_event('WebSocket closed. Cannot write.')
goto free_data unsafe {
free(payload)
}
return -1 return -1
} }
header_len := 6 + if payload_len > 125 { 2 } else { 0 } + if payload_len > 0xffff { 6 } else { 0 } header_len := 6 + if payload_len > 125 { 2 } else { 0 } + if payload_len > 0xffff { 6 } else { 0 }
frame_len := header_len + payload_len
mut frame_buf := [`0`].repeat(frame_len)
fbdata := byteptr( frame_buf.data )
masking_key := create_masking_key() masking_key := create_masking_key()
mut header := [`0`].repeat(header_len) mut header := [`0`].repeat(header_len)
mut bytes_written := -1
header[0] = (int(code) | 0x80) header[0] = (int(code) | 0x80)
if payload_len <= 125 { if payload_len <= 125 {
header[1] = (payload_len | 0x80) header[1] = (payload_len | 0x80)
@ -284,31 +272,25 @@ pub fn (mut ws Client) write(payload byteptr, payload_len int, code OPCode) int
header[12] = masking_key[2] header[12] = masking_key[2]
header[13] = masking_key[3] header[13] = masking_key[3]
} else { } else {
l.c("write: frame too large") l.c('write: frame too large')
ws.close(1009, "frame too large") ws.close(1009, 'frame too large')
goto free_data goto free_data
return -1 return -1
} }
C.memcpy(fbdata, header.data, header_len)
frame_len := header_len + payload_len C.memcpy(fbdata + header_len, payload, payload_len)
mut frame_buf := [`0`].repeat(frame_len)
C.memcpy(frame_buf.data, header.data, header_len)
C.memcpy(&frame_buf.data[header_len], payload, payload_len)
for i in 0 .. payload_len { for i in 0 .. payload_len {
frame_buf[header_len + i] ^= masking_key[i % 4] & 0xff frame_buf[header_len + i] ^= masking_key[i % 4] & 0xff
} }
bytes_written = ws.write_to_server(fbdata, frame_len)
bytes_written = ws.write_to_server(frame_buf.data, frame_len)
if bytes_written == -1 { if bytes_written == -1 {
err := string(byteptr(C.strerror(C.errno))) err := string(byteptr(C.strerror(C.errno)))
l.e("write: there was an error writing data: ${err}") l.e('write: there was an error writing data: ${err}')
ws.send_error_event("Error writing data") ws.send_error_event('Error writing data')
goto free_data goto free_data
return -1 return -1
} }
l.d("write: ${bytes_written} bytes written.") l.d('write: ${bytes_written} bytes written.')
free_data: free_data:
unsafe { unsafe {
free(payload) free(payload)
@ -320,34 +302,30 @@ pub fn (mut ws Client) write(payload byteptr, payload_len int, code OPCode) int
} }
pub fn (mut ws Client) listen() { pub fn (mut ws Client) listen() {
l.i("Starting listener...") l.i('Starting listener...')
for ws.state == .open { for ws.state == .open {
ws.read() ws.read()
} }
l.i("Listener stopped as websocket was closed.") l.i('Listener stopped as websocket was closed.')
} }
pub fn (mut ws Client) read() int { pub fn (mut ws Client) read() int {
mut bytes_read := u64(0) mut bytes_read := u64(0)
initial_buffer := u64(256) initial_buffer := u64(256)
mut header_len := 2 mut header_len := 2
header_len_offset := 2 header_len_offset := 2
extended_payload16_end_byte := 4 extended_payload16_end_byte := 4
extended_payload64_end_byte := 10 extended_payload64_end_byte := 10
mut payload_len := u64(0) mut payload_len := u64(0)
mut data := C.calloc(initial_buffer, 1) // [`0`].repeat(int(max_buffer)) mut data := C.calloc(initial_buffer, 1) // [`0`].repeat(int(max_buffer))
mut frame := Frame{} mut frame := Frame{}
mut frame_size := u64(header_len) mut frame_size := u64(header_len)
for bytes_read < frame_size && ws.state == .open { for bytes_read < frame_size && ws.state == .open {
byt := ws.read_from_server(data + int(bytes_read), 1) byt := ws.read_from_server(data + int(bytes_read), 1)
match byt { match byt {
0 { 0 {
error := "server closed the connection." error := 'server closed the connection.'
l.e("read: ${error}") l.e('read: ${error}')
ws.send_error_event(error) ws.send_error_event(error)
ws.close(1006, error) ws.close(1006, error)
goto free_data goto free_data
@ -355,11 +333,12 @@ pub fn (mut ws Client) read() int {
} }
-1 { -1 {
err := string(byteptr(C.strerror(C.errno))) err := string(byteptr(C.strerror(C.errno)))
l.e("read: error reading frame. ${err}") l.e('read: error reading frame. ${err}')
ws.send_error_event("error reading frame") ws.send_error_event('error reading frame')
goto free_data goto free_data
return -1 return -1
} else { }
else {
bytes_read++ bytes_read++
} }
} }
@ -371,7 +350,6 @@ pub fn (mut ws Client) read() int {
frame.opcode = OPCode(int(data[0] & 0x7F)) frame.opcode = OPCode(int(data[0] & 0x7F))
frame.mask = (data[1] & 0x80) == 0x80 frame.mask = (data[1] & 0x80) == 0x80
frame.payload_len = u64(data[1] & 0x7F) frame.payload_len = u64(data[1] & 0x7F)
// masking key // masking key
if frame.mask { if frame.mask {
frame.masking_key[0] = data[2] frame.masking_key[0] = data[2]
@ -379,18 +357,14 @@ pub fn (mut ws Client) read() int {
frame.masking_key[2] = data[4] frame.masking_key[2] = data[4]
frame.masking_key[3] = data[5] frame.masking_key[3] = data[5]
} }
payload_len = frame.payload_len payload_len = frame.payload_len
frame_size = u64(header_len) + payload_len frame_size = u64(header_len) + payload_len
} }
if frame.payload_len == u64(126) && bytes_read == u64(extended_payload16_end_byte) { if frame.payload_len == u64(126) && bytes_read == u64(extended_payload16_end_byte) {
header_len += 2 header_len += 2
mut extended_payload_len := 0 mut extended_payload_len := 0
extended_payload_len |= data[2] << 8 extended_payload_len |= data[2] << 8
extended_payload_len |= data[3] << 0 extended_payload_len |= data[3] << 0
// masking key // masking key
if frame.mask { if frame.mask {
frame.masking_key[0] = data[4] frame.masking_key[0] = data[4]
@ -398,16 +372,14 @@ pub fn (mut ws Client) read() int {
frame.masking_key[2] = data[6] frame.masking_key[2] = data[6]
frame.masking_key[3] = data[7] frame.masking_key[3] = data[7]
} }
payload_len = u64(extended_payload_len) payload_len = u64(extended_payload_len)
frame_size = u64(header_len) + payload_len frame_size = u64(header_len) + payload_len
if frame_size > initial_buffer { if frame_size > initial_buffer {
l.d("reallocating: ${frame_size}") l.d('reallocating: ${frame_size}')
data = C.realloc(data, frame_size) data = C.realloc(data, frame_size)
} }
} else if frame.payload_len == u64(127) && bytes_read == u64(extended_payload64_end_byte) { } else if frame.payload_len == u64(127) && bytes_read == u64(extended_payload64_end_byte) {
header_len += 8 // TODO Not sure... header_len += 8 // TODO Not sure...
mut extended_payload_len := u64(0) mut extended_payload_len := u64(0)
extended_payload_len |= u64(data[2]) << 56 extended_payload_len |= u64(data[2]) << 56
extended_payload_len |= u64(data[3]) << 48 extended_payload_len |= u64(data[3]) << 48
@ -417,7 +389,6 @@ pub fn (mut ws Client) read() int {
extended_payload_len |= u64(data[7]) << 16 extended_payload_len |= u64(data[7]) << 16
extended_payload_len |= u64(data[8]) << 8 extended_payload_len |= u64(data[8]) << 8
extended_payload_len |= u64(data[9]) << 0 extended_payload_len |= u64(data[9]) << 0
// masking key // masking key
if frame.mask { if frame.mask {
frame.masking_key[0] = data[10] frame.masking_key[0] = data[10]
@ -425,33 +396,30 @@ pub fn (mut ws Client) read() int {
frame.masking_key[2] = data[12] frame.masking_key[2] = data[12]
frame.masking_key[3] = data[13] frame.masking_key[3] = data[13]
} }
payload_len = extended_payload_len payload_len = extended_payload_len
frame_size = u64(header_len) + payload_len frame_size = u64(header_len) + payload_len
if frame_size > initial_buffer { if frame_size > initial_buffer {
l.d("reallocating: ${frame_size}") l.d('reallocating: ${frame_size}')
data = C.realloc(data, frame_size) data = C.realloc(data, frame_size)
} }
} }
} }
// unmask the payload // unmask the payload
if frame.mask { if frame.mask {
for i in 0 .. payload_len { for i in 0 .. payload_len {
data[header_len + i] ^= frame.masking_key[i % 4] & 0xff data[header_len + i] ^= frame.masking_key[i % 4] & 0xff
} }
} }
if ws.fragments.len > 0 && frame.opcode in [.text_frame, .binary_frame] { if ws.fragments.len > 0 && frame.opcode in [.text_frame, .binary_frame] {
ws.close(0, "") ws.close(0, '')
goto free_data goto free_data
return -1 return -1
} else if frame.opcode in [.text_frame, .binary_frame] { } else if frame.opcode in [.text_frame, .binary_frame] {
data_node: data_node:
l.d("read: recieved text_frame or binary_frame") l.d('read: recieved text_frame or binary_frame')
mut payload := malloc(sizeof(byte) * int(payload_len) + 1) mut payload := malloc(sizeof(byte) * int(payload_len) + 1)
if payload == C.NULL { if payload == 0 {
l.f("out of memory") l.f('out of memory')
} }
C.memcpy(payload, &data[header_len], payload_len) C.memcpy(payload, &data[header_len], payload_len)
if frame.fin { if frame.fin {
@ -470,14 +438,16 @@ pub fn (mut ws Client) read() int {
} }
} }
mut pl := malloc(sizeof(byte) * int(size)) mut pl := malloc(sizeof(byte) * int(size))
if pl == C.NULL { if pl == 0 {
l.f("out of memory") l.f('out of memory')
} }
mut by := 0 mut by := 0
for i, f in frags { for f in frags {
C.memcpy(pl + by, f.data, f.len) C.memcpy(pl + by, f.data, f.len)
by += int(f.len) by += int(f.len)
unsafe {free(f.data)} unsafe {
free(f.data)
}
} }
payload = pl payload = pl
frame.opcode = ws.fragments[0].code frame.opcode = ws.fragments[0].code
@ -491,9 +461,9 @@ pub fn (mut ws Client) read() int {
payload[payload_len] = `\0` payload[payload_len] = `\0`
if frame.opcode == .text_frame && payload_len > 0 { if frame.opcode == .text_frame && payload_len > 0 {
if !utf8_validate(payload, int(payload_len)) { if !utf8_validate(payload, int(payload_len)) {
l.e("malformed utf8 payload") l.e('malformed utf8 payload')
ws.send_error_event("Recieved malformed utf8.") ws.send_error_event('Recieved malformed utf8.')
ws.close(1007, "malformed utf8 payload") ws.close(1007, 'malformed utf8 payload')
goto free_data goto free_data
return -1 return -1
} }
@ -516,27 +486,25 @@ pub fn (mut ws Client) read() int {
free(data) free(data)
} }
return int(bytes_read) return int(bytes_read)
} } else if frame.opcode == .continuation {
else if frame.opcode == .continuation { l.d('read: continuation')
l.d("read: continuation")
if ws.fragments.len <= 0 { if ws.fragments.len <= 0 {
l.e("Nothing to continue.") l.e('Nothing to continue.')
ws.close(1002, "nothing to continue") ws.close(1002, 'nothing to continue')
goto free_data goto free_data
return -1 return -1
} }
goto data_node goto data_node
return 0 return 0
} } else if frame.opcode == .ping {
else if frame.opcode == .ping { l.d('read: ping')
l.d("read: ping")
if !frame.fin { if !frame.fin {
ws.close(1002, "control message must not be fragmented") ws.close(1002, 'control message must not be fragmented')
goto free_data goto free_data
return -1 return -1
} }
if frame.payload_len > 125 { if frame.payload_len > 125 {
ws.close(1002, "control frames must not exceed 125 bytes") ws.close(1002, 'control frames must not exceed 125 bytes')
goto free_data goto free_data
return -1 return -1
} }
@ -548,11 +516,10 @@ pub fn (mut ws Client) read() int {
unsafe { unsafe {
free(data) free(data)
} }
return ws.send_control_frame(.pong, "PONG", payload) return ws.send_control_frame(.pong, 'PONG', payload)
} } else if frame.opcode == .pong {
else if frame.opcode == .pong {
if !frame.fin { if !frame.fin {
ws.close(1002, "control message must not be fragmented") ws.close(1002, 'control message must not be fragmented')
goto free_data goto free_data
return -1 return -1
} }
@ -561,26 +528,25 @@ pub fn (mut ws Client) read() int {
} }
// got pong // got pong
return 0 return 0
} } else if frame.opcode == .close {
else if frame.opcode == .close { l.d('read: close')
l.d("read: close")
if frame.payload_len > 125 { if frame.payload_len > 125 {
ws.close(1002, "control frames must not exceed 125 bytes") ws.close(1002, 'control frames must not exceed 125 bytes')
goto free_data goto free_data
return -1 return -1
} }
mut code := 0 mut code := 0
mut reason := "" mut reason := ''
if payload_len > 2 { if payload_len > 2 {
code = (int(data[header_len]) << 8) + int(data[header_len + 1]) code = (int(data[header_len]) << 8) + int(data[header_len + 1])
header_len += 2 header_len += 2
payload_len -= 2 payload_len -= 2
reason = string(&data[header_len]) reason = string(&data[header_len])
l.i("Closing with reason: ${reason} & code: ${code}") l.i('Closing with reason: ${reason} & code: ${code}')
if reason.len > 1 && !utf8_validate(reason.str, reason.len) { if reason.len > 1 && !utf8_validate(reason.str, reason.len) {
l.e("malformed utf8 payload") l.e('malformed utf8 payload')
ws.send_error_event("Recieved malformed utf8.") ws.send_error_event('Recieved malformed utf8.')
ws.close(1007, "malformed utf8 payload") ws.close(1007, 'malformed utf8 payload')
goto free_data goto free_data
return -1 return -1
} }
@ -591,9 +557,9 @@ pub fn (mut ws Client) read() int {
ws.close(code, reason) ws.close(code, reason)
return 0 return 0
} }
l.e("read: Recieved unsupported opcode: ${frame.opcode} fin: ${frame.fin} uri: ${ws.uri}") l.e('read: Recieved unsupported opcode: ${frame.opcode} fin: ${frame.fin} uri: ${ws.uri}')
ws.send_error_event("Recieved unsupported opcode: ${frame.opcode}") ws.send_error_event('Recieved unsupported opcode: ${frame.opcode}')
ws.close(1002, "Unsupported opcode") ws.close(1002, 'Unsupported opcode')
free_data: free_data:
unsafe { unsafe {
free(data) free(data)
@ -602,9 +568,13 @@ pub fn (mut ws Client) read() int {
} }
fn (mut ws Client) send_control_frame(code OPCode, frame_typ string, payload []byte) int { fn (mut ws Client) send_control_frame(code OPCode, frame_typ string, payload []byte) int {
mut bytes_written := -1
if ws.socket.sockfd <= 0 { if ws.socket.sockfd <= 0 {
l.e("No socket opened.") l.e('No socket opened.')
goto free_data unsafe {
payload.free()
}
l.c('send_control_frame: error sending ${frame_typ} control frame.')
return -1 return -1
} }
header_len := 6 header_len := 6
@ -618,7 +588,6 @@ fn (mut ws Client) send_control_frame(code OPCode, frame_typ string, payload []b
control_frame[4] = masking_key[2] control_frame[4] = masking_key[2]
control_frame[5] = masking_key[3] control_frame[5] = masking_key[3]
if code == .close { if code == .close {
close_code := 1000
if payload.len > 2 { if payload.len > 2 {
mut parsed_payload := [`0`].repeat(payload.len + 1) mut parsed_payload := [`0`].repeat(payload.len + 1)
C.memcpy(parsed_payload.data, &payload[0], payload.len) C.memcpy(parsed_payload.data, &payload[0], payload.len)
@ -635,7 +604,6 @@ fn (mut ws Client) send_control_frame(code OPCode, frame_typ string, payload []b
control_frame[header_len + i] = (payload[i] ^ masking_key[i % 4]) & 0xff control_frame[header_len + i] = (payload[i] ^ masking_key[i % 4]) & 0xff
} }
} }
mut bytes_written := -1
bytes_written = ws.write_to_server(control_frame.data, frame_len) bytes_written = ws.write_to_server(control_frame.data, frame_len)
free_data: free_data:
unsafe { unsafe {
@ -645,14 +613,15 @@ fn (mut ws Client) send_control_frame(code OPCode, frame_typ string, payload []b
} }
match bytes_written { match bytes_written {
0 { 0 {
l.d("send_control_frame: remote host closed the connection.") l.d('send_control_frame: remote host closed the connection.')
return 0 return 0
} }
-1 { -1 {
l.c("send_control_frame: error sending ${frame_typ} control frame.") l.c('send_control_frame: error sending ${frame_typ} control frame.')
return -1 return -1
} else { }
l.d("send_control_frame: wrote ${bytes_written} byte ${frame_typ} frame.") else {
l.d('send_control_frame: wrote ${bytes_written} byte ${frame_typ} frame.')
return bytes_written return bytes_written
} }
} }

6
vlib/net/websocket/ws_test.v 100644
View File

@ -0,0 +1,6 @@
module websocket
// TODO This only checks that the client compiles, do real tests
fn test_compile() {
new('http://examples.com')
}