// // Copyright 2012 Square Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. // #import #import "FSRWebSocket.h" #if TARGET_OS_IOS || TARGET_OS_TV #define HAS_ICU #endif #import #ifdef HAS_ICU #import #endif #if TARGET_OS_IOS || TARGET_OS_TV #import #elif TARGET_OS_OSX #import #endif #import #import #import "fbase64.h" #import "NSData+SRB64Additions.h" #if OS_OBJECT_USE_OBJC_RETAIN_RELEASE #define sr_dispatch_retain(x) #define sr_dispatch_release(x) #define maybe_bridge(x) ((__bridge void *) x) #else #define sr_dispatch_retain(x) dispatch_retain(x) #define sr_dispatch_release(x) dispatch_release(x) #define maybe_bridge(x) (x) #endif typedef enum { SROpCodeTextFrame = 0x1, SROpCodeBinaryFrame = 0x2, //3-7Reserved SROpCodeConnectionClose = 0x8, SROpCodePing = 0x9, SROpCodePong = 0xA, //B-F reserved } FSROpCode; typedef enum { SRStatusCodeNormal = 1000, SRStatusCodeGoingAway = 1001, SRStatusCodeProtocolError = 1002, SRStatusCodeUnhandledType = 1003, // 1004 reserved SRStatusNoStatusReceived = 1005, // 1004-1006 reserved SRStatusCodeInvalidUTF8 = 1007, SRStatusCodePolicyViolated = 1008, SRStatusCodeMessageTooBig = 1009, } FSRStatusCode; typedef struct { BOOL fin; // BOOL rsv1; // BOOL rsv2; // BOOL rsv3; uint8_t opcode; BOOL masked; uint64_t payload_length; } frame_header; static NSString *const SRWebSocketAppendToSecKeyString = @"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; static inline int32_t validate_dispatch_data_partial_string(NSData *data); static inline void SRFastLog(NSString *format, ...); @interface NSData (FSRWebSocket) - (NSString *)stringBySHA1ThenBase64Encoding; @end @interface NSString (FSRWebSocket) - (NSString *)stringBySHA1ThenBase64Encoding; @end @interface NSURL (FSRWebSocket) // The origin isn't really applicable for a native application // So instead, just map ws -> http and wss -> https - (NSString *)SR_origin; @end @interface _FSRRunLoopThread : NSThread @property (nonatomic, readonly) NSRunLoop *runLoop; @end static NSString *newSHA1String(const char *bytes, size_t length) { uint8_t md[CC_SHA1_DIGEST_LENGTH]; CC_SHA1(bytes, (int)length, md); size_t buffer_size = ((sizeof(md) * 3 + 2) / 2); char *buffer = (char *)malloc(buffer_size); int len = f_b64_ntop(md, CC_SHA1_DIGEST_LENGTH, buffer, buffer_size); if (len == -1) { free(buffer); return nil; } else{ return [[NSString alloc] initWithBytesNoCopy:buffer length:len encoding:NSASCIIStringEncoding freeWhenDone:YES]; } } @implementation NSData (FSRWebSocket) - (NSString *)stringBySHA1ThenBase64Encoding; { return newSHA1String(self.bytes, self.length); } @end @implementation NSString (FSRWebSocket) - (NSString *)stringBySHA1ThenBase64Encoding; { return newSHA1String(self.UTF8String, self.length); } @end NSString *const FSRWebSocketErrorDomain = @"FSRWebSocketErrorDomain"; // Returns number of bytes consumed. returning 0 means you didn't match. // Sends bytes to callback handler; typedef size_t (^stream_scanner)(NSData *collected_data); typedef void (^data_callback)(FSRWebSocket *webSocket, NSData *data); @interface FSRIOConsumer : NSObject { stream_scanner _scanner; data_callback _handler; size_t _bytesNeeded; BOOL _readToCurrentFrame; BOOL _unmaskBytes; } @property (nonatomic, copy, readonly) stream_scanner consumer; @property (nonatomic, copy, readonly) data_callback handler; @property (nonatomic, assign) size_t bytesNeeded; @property (nonatomic, assign, readonly) BOOL readToCurrentFrame; @property (nonatomic, assign, readonly) BOOL unmaskBytes; @end // This class is not thread-safe, and is expected to always be run on the same queue. @interface FSRIOConsumerPool : NSObject - (id)initWithBufferCapacity:(NSUInteger)poolSize; - (FSRIOConsumer *)consumerWithScanner:(stream_scanner)scanner handler:(data_callback)handler bytesNeeded:(size_t)bytesNeeded readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes; - (void)returnConsumer:(FSRIOConsumer *)consumer; @end @interface FSRWebSocket () - (void)_writeData:(NSData *)data; - (void)_closeWithProtocolError:(NSString *)message; - (void)_failWithError:(NSError *)error; - (void)_disconnect; - (void)_readFrameNew; - (void)_readFrameContinue; - (void)_pumpScanner; - (void)_pumpWriting; - (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback; - (void)_addConsumerWithDataLength:(size_t)dataLength callback:(data_callback)callback readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes; - (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback dataLength:(size_t)dataLength; - (void)_readUntilBytes:(const void *)bytes length:(size_t)length callback:(data_callback)dataHandler; - (void)_readUntilHeaderCompleteWithCallback:(data_callback)dataHandler; - (void)_sendFrameWithOpcode:(FSROpCode)opcode data:(id)data; - (BOOL)_checkHandshake:(CFHTTPMessageRef)httpMessage; - (void)_SR_commonInit; - (void)_initializeStreams; - (void)_connect; @property (nonatomic) FSRReadyState readyState; @property (nonatomic) NSOperationQueue *delegateOperationQueue; @property (nonatomic) dispatch_queue_t delegateDispatchQueue; @end @implementation FSRWebSocket { NSInteger _webSocketVersion; NSOperationQueue *_delegateOperationQueue; dispatch_queue_t _delegateDispatchQueue; dispatch_queue_t _workQueue; NSMutableArray *_consumers; NSInputStream *_inputStream; NSOutputStream *_outputStream; NSMutableData *_readBuffer; NSInteger _readBufferOffset; NSMutableData *_outputBuffer; NSInteger _outputBufferOffset; uint8_t _currentFrameOpcode; size_t _currentFrameCount; size_t _readOpCount; uint32_t _currentStringScanPosition; NSMutableData *_currentFrameData; NSString *_closeReason; NSString *_secKey; BOOL _pinnedCertFound; uint8_t _currentReadMaskKey[4]; size_t _currentReadMaskOffset; BOOL _consumerStopped; BOOL _closeWhenFinishedWriting; BOOL _failed; BOOL _secure; NSURLRequest *_urlRequest; NSString *_userAgent; CFHTTPMessageRef _receivedHTTPHeaders; BOOL _sentClose; BOOL _didFail; BOOL _cleanupScheduled; int _closeCode; BOOL _isPumping; NSMutableSet *_scheduledRunloops; // We use this to retain ourselves. __strong FSRWebSocket *_selfRetain; NSArray *_requestedProtocols; FSRIOConsumerPool *_consumerPool; } @synthesize delegate = _delegate; @synthesize url = _url; @synthesize readyState = _readyState; @synthesize protocol = _protocol; static __strong NSData *CRLFCRLF; + (void)initialize; { CRLFCRLF = [[NSData alloc] initWithBytes:"\r\n\r\n" length:4]; } - (id)initWithURLRequest:(NSURLRequest *)request protocols:(NSArray *)protocols queue:(dispatch_queue_t)queue andUserAgent:(NSString *)userAgent; { self = [super init]; if (self) { assert(request.URL); _url = request.URL; NSString *scheme = [_url scheme]; _requestedProtocols = [protocols copy]; _userAgent = userAgent; assert([scheme isEqualToString:@"ws"] || [scheme isEqualToString:@"http"] || [scheme isEqualToString:@"wss"] || [scheme isEqualToString:@"https"]); _urlRequest = request; if ([scheme isEqualToString:@"wss"] || [scheme isEqualToString:@"https"]) { _secure = YES; } if (!queue) { _delegateDispatchQueue = dispatch_get_main_queue(); } else { _delegateDispatchQueue = queue; } [self _SR_commonInit]; } return self; } - (id)initWithURLRequest:(NSURLRequest *)request protocols:(NSArray *)protocols; { return [self initWithURLRequest:request protocols:nil queue:nil andUserAgent:nil]; } - (id)initWithURLRequest:(NSURLRequest *)request queue:(dispatch_queue_t)queue andUserAgent:(NSString *)userAgent; { return [self initWithURLRequest:request protocols:nil queue:queue andUserAgent:userAgent]; } - (id)initWithURLRequest:(NSURLRequest *)request; { return [self initWithURLRequest:request protocols:nil]; } - (id)initWithURL:(NSURL *)url; { return [self initWithURL:url protocols:nil]; } - (id)initWithURL:(NSURL *)url protocols:(NSArray *)protocols; { NSMutableURLRequest *request = [[NSMutableURLRequest alloc] initWithURL:url]; return [self initWithURLRequest:request protocols:protocols]; } - (void)_SR_commonInit; { _readyState = SR_CONNECTING; _consumerStopped = YES; _webSocketVersion = 13; _workQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL); // Going to set a specific on the queue so we can validate we're on the work queue dispatch_queue_set_specific(_workQueue, (__bridge void *)self, maybe_bridge(_workQueue), NULL); sr_dispatch_retain(_delegateDispatchQueue); _readBuffer = [[NSMutableData alloc] init]; _outputBuffer = [[NSMutableData alloc] init]; _currentFrameData = [[NSMutableData alloc] init]; _consumers = [[NSMutableArray alloc] init]; _consumerPool = [[FSRIOConsumerPool alloc] init]; _scheduledRunloops = [[NSMutableSet alloc] init]; [self _initializeStreams]; // default handlers } - (void)assertOnWorkQueue; { assert(dispatch_get_specific((__bridge void *)self) == maybe_bridge(_workQueue)); } - (void)dealloc { _inputStream.delegate = nil; _outputStream.delegate = nil; [_inputStream close]; [_outputStream close]; sr_dispatch_release(_workQueue); _workQueue = NULL; if (_receivedHTTPHeaders) { CFRelease(_receivedHTTPHeaders); _receivedHTTPHeaders = NULL; } if (_delegateDispatchQueue) { sr_dispatch_release(_delegateDispatchQueue); _delegateDispatchQueue = NULL; } } #ifndef NDEBUG - (void)setReadyState:(FSRReadyState)aReadyState; { [self willChangeValueForKey:@"readyState"]; assert(aReadyState > _readyState); _readyState = aReadyState; [self didChangeValueForKey:@"readyState"]; } #endif - (void)open; { assert(_url); NSAssert(_readyState == SR_CONNECTING, @"Cannot call -(void)open on SRWebSocket more than once"); _selfRetain = self; [self _connect]; } // Calls block on delegate queue - (void)_performDelegateBlock:(dispatch_block_t)block; { if (_delegateOperationQueue) { [_delegateOperationQueue addOperationWithBlock:block]; } else { assert(_delegateDispatchQueue); dispatch_async(_delegateDispatchQueue, block); } } - (void)setDelegateDispatchQueue:(dispatch_queue_t)queue; { if (queue) { sr_dispatch_retain(queue); } if (_delegateDispatchQueue) { sr_dispatch_release(_delegateDispatchQueue); } _delegateDispatchQueue = queue; } - (BOOL)_checkHandshake:(CFHTTPMessageRef)httpMessage; { NSString *acceptHeader = CFBridgingRelease(CFHTTPMessageCopyHeaderFieldValue(httpMessage, CFSTR("Sec-WebSocket-Accept"))); if (acceptHeader == nil) { return NO; } NSString *concattedString = [_secKey stringByAppendingString:SRWebSocketAppendToSecKeyString]; NSString *expectedAccept = [concattedString stringBySHA1ThenBase64Encoding]; return [acceptHeader isEqualToString:expectedAccept]; } - (void)_HTTPHeadersDidFinish; { NSInteger responseCode = CFHTTPMessageGetResponseStatusCode(_receivedHTTPHeaders); if (responseCode >= 400) { SRFastLog(@"Request failed with response code %d", responseCode); [self _failWithError:[NSError errorWithDomain:@"org.lolrus.SocketRocket" code:2132 userInfo:[NSDictionary dictionaryWithObject:[NSString stringWithFormat:@"received bad response code from server %u", (int)responseCode] forKey:NSLocalizedDescriptionKey]]]; return; } if(![self _checkHandshake:_receivedHTTPHeaders]) { [self _failWithError:[NSError errorWithDomain:FSRWebSocketErrorDomain code:2133 userInfo:[NSDictionary dictionaryWithObject:[NSString stringWithFormat:@"Invalid Sec-WebSocket-Accept response"] forKey:NSLocalizedDescriptionKey]]]; return; } NSString *negotiatedProtocol = CFBridgingRelease(CFHTTPMessageCopyHeaderFieldValue(_receivedHTTPHeaders, CFSTR("Sec-WebSocket-Protocol"))); if (negotiatedProtocol) { // Make sure we requested the protocol if ([_requestedProtocols indexOfObject:negotiatedProtocol] == NSNotFound) { [self _failWithError:[NSError errorWithDomain:FSRWebSocketErrorDomain code:2133 userInfo:[NSDictionary dictionaryWithObject:[NSString stringWithFormat:@"Server specified Sec-WebSocket-Protocol that wasn't requested"] forKey:NSLocalizedDescriptionKey]]]; return; } _protocol = negotiatedProtocol; } self.readyState = SR_OPEN; if (!_didFail) { [self _readFrameNew]; } [self _performDelegateBlock:^{ if ([self.delegate respondsToSelector:@selector(webSocketDidOpen:)]) { [self.delegate webSocketDidOpen:self]; }; }]; } - (void)_readHTTPHeader; { if (_receivedHTTPHeaders == NULL) { _receivedHTTPHeaders = CFHTTPMessageCreateEmpty(NULL, NO); } [self _readUntilHeaderCompleteWithCallback:^(FSRWebSocket *self, NSData *data) { CFHTTPMessageAppendBytes(_receivedHTTPHeaders, (const UInt8 *)data.bytes, data.length); if (CFHTTPMessageIsHeaderComplete(_receivedHTTPHeaders)) { SRFastLog(@"Finished reading headers %@", CFBridgingRelease(CFHTTPMessageCopyAllHeaderFields(_receivedHTTPHeaders))); [self _HTTPHeadersDidFinish]; } else { [self _readHTTPHeader]; } }]; } - (void)didConnect { SRFastLog(@"Connected"); CFHTTPMessageRef request = CFHTTPMessageCreateRequest(NULL, CFSTR("GET"), (__bridge CFURLRef)_url, kCFHTTPVersion1_1); // Set host first so it defaults CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Host"), (__bridge CFStringRef)(_url.port ? [NSString stringWithFormat:@"%@:%@", _url.host, _url.port] : _url.host)); NSMutableData *keyBytes = [[NSMutableData alloc] initWithLength:16]; int result = SecRandomCopyBytes(kSecRandomDefault, keyBytes.length, keyBytes.mutableBytes); assert(result == 0); _secKey = [FSRUtilities base64EncodedStringFromData:keyBytes]; assert([_secKey length] == 24); CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Upgrade"), CFSTR("websocket")); CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Connection"), CFSTR("Upgrade")); if (_userAgent) { CFHTTPMessageSetHeaderFieldValue(request, CFSTR("User-Agent"), (__bridge CFStringRef)_userAgent); } CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Sec-WebSocket-Key"), (__bridge CFStringRef)_secKey); CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Sec-WebSocket-Version"), (__bridge CFStringRef)[NSString stringWithFormat:@"%u", (int)_webSocketVersion]); CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Origin"), (__bridge CFStringRef)_url.SR_origin); if (_requestedProtocols) { CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Sec-WebSocket-Protocol"), (__bridge CFStringRef)[_requestedProtocols componentsJoinedByString:@", "]); } [_urlRequest.allHTTPHeaderFields enumerateKeysAndObjectsUsingBlock:^(id key, id obj, BOOL *stop) { CFHTTPMessageSetHeaderFieldValue(request, (__bridge CFStringRef)key, (__bridge CFStringRef)obj); }]; NSData *message = CFBridgingRelease(CFHTTPMessageCopySerializedMessage(request)); CFRelease(request); [self _writeData:message]; [self _readHTTPHeader]; } //- (void)_connectToHost:(NSString *)host port:(NSInteger)port; - (void)_initializeStreams; { NSInteger port = _url.port.integerValue; if (port == 0) { if (!_secure) { port = 80; } else { port = 443; } } NSString *host = _url.host; CFReadStreamRef readStream = NULL; CFWriteStreamRef writeStream = NULL; CFStreamCreatePairWithSocketToHost(NULL, (__bridge CFStringRef)host, (int)port, &readStream, &writeStream); // XXX CFReadStreamSetProperty(readStream, kCFStreamNetworkServiceType, kCFStreamNetworkServiceTypeBackground); CFWriteStreamSetProperty(writeStream, kCFStreamNetworkServiceType, kCFStreamNetworkServiceTypeBackground); _outputStream = CFBridgingRelease(writeStream); _inputStream = CFBridgingRelease(readStream); if (_secure) { NSMutableDictionary *SSLOptions = [[NSMutableDictionary alloc] init]; [_outputStream setProperty:(__bridge id)kCFStreamSocketSecurityLevelNegotiatedSSL forKey:(__bridge id)kCFStreamPropertySocketSecurityLevel]; // If we're using pinned certs, don't validate the certificate chain if ([_urlRequest FSR_SSLPinnedCertificates].count) { [SSLOptions setValue:[NSNumber numberWithBool:NO] forKey:(__bridge id)kCFStreamSSLValidatesCertificateChain]; } [_outputStream setProperty:SSLOptions forKey:(__bridge id)kCFStreamPropertySSLSettings]; } _inputStream.delegate = self; _outputStream.delegate = self; [_outputStream open]; [_inputStream open]; } - (void)_connect; { if (!_scheduledRunloops.count) { [self scheduleInRunLoop:[NSRunLoop FSR_networkRunLoop] forMode:NSDefaultRunLoopMode]; } [_outputStream open]; [_inputStream open]; } - (void)scheduleInRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode; { [_outputStream scheduleInRunLoop:aRunLoop forMode:mode]; [_inputStream scheduleInRunLoop:aRunLoop forMode:mode]; [_scheduledRunloops addObject:@[aRunLoop, mode]]; } - (void)unscheduleFromRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode; { [_outputStream removeFromRunLoop:aRunLoop forMode:mode]; [_inputStream removeFromRunLoop:aRunLoop forMode:mode]; [_scheduledRunloops removeObject:@[aRunLoop, mode]]; } - (void)close; { [self closeWithCode:-1 reason:nil]; } - (void)closeWithCode:(NSInteger)code reason:(NSString *)reason; { assert(code); dispatch_async(_workQueue, ^{ if (self.readyState == SR_CLOSING || self.readyState == SR_CLOSED) { return; } BOOL wasConnecting = self.readyState == SR_CONNECTING; self.readyState = SR_CLOSING; SRFastLog(@"Closing with code %d reason %@", code, reason); if (wasConnecting) { [self _disconnect]; return; } size_t maxMsgSize = [reason maximumLengthOfBytesUsingEncoding:NSUTF8StringEncoding]; NSMutableData *mutablePayload = [[NSMutableData alloc] initWithLength:sizeof(uint16_t) + maxMsgSize]; NSData *payload = mutablePayload; ((uint16_t *)mutablePayload.mutableBytes)[0] = EndianU16_BtoN(code); if (reason) { NSRange remainingRange = {0}; NSUInteger usedLength = 0; BOOL success = [reason getBytes:(char *)mutablePayload.mutableBytes + sizeof(uint16_t) maxLength:payload.length - sizeof(uint16_t) usedLength:&usedLength encoding:NSUTF8StringEncoding options:NSStringEncodingConversionExternalRepresentation range:NSMakeRange(0, reason.length) remainingRange:&remainingRange]; assert(success); assert(remainingRange.length == 0); if (usedLength != maxMsgSize) { payload = [payload subdataWithRange:NSMakeRange(0, usedLength + sizeof(uint16_t))]; } } [self _sendFrameWithOpcode:SROpCodeConnectionClose data:payload]; }); } - (void)_closeWithProtocolError:(NSString *)message; { // Need to shunt this on the _callbackQueue first to see if they received any messages [self _performDelegateBlock:^{ [self closeWithCode:SRStatusCodeProtocolError reason:message]; dispatch_async(_workQueue, ^{ [self _disconnect]; }); }]; } - (void)_failWithError:(NSError *)error; { dispatch_async(_workQueue, ^{ if (self.readyState != SR_CLOSED) { _failed = YES; [self _performDelegateBlock:^{ if ([self.delegate respondsToSelector:@selector(webSocket:didFailWithError:)]) { [self.delegate webSocket:self didFailWithError:error]; } }]; self.readyState = SR_CLOSED; SRFastLog(@"Failing with error %@", error.localizedDescription); [self _disconnect]; [self _scheduleCleanup]; } }); } - (void)_writeData:(NSData *)data; { [self assertOnWorkQueue]; if (_closeWhenFinishedWriting) { return; } [_outputBuffer appendData:data]; [self _pumpWriting]; } - (void)send:(id)data; { SRFastLog(@"Sending data %@", data); NSAssert(self.readyState != SR_CONNECTING, @"Invalid State: Cannot call send: until connection is open"); // TODO: maybe not copy this for performance data = [data copy]; dispatch_async(_workQueue, ^{ if ([data isKindOfClass:[NSString class]]) { [self _sendFrameWithOpcode:SROpCodeTextFrame data:[(NSString *)data dataUsingEncoding:NSUTF8StringEncoding]]; } else if ([data isKindOfClass:[NSData class]]) { [self _sendFrameWithOpcode:SROpCodeBinaryFrame data:data]; } else if (data == nil) { [self _sendFrameWithOpcode:SROpCodeTextFrame data:data]; } else { assert(NO); } }); } - (void)handlePing:(NSData *)pingData; { // Need to pingpong this off _callbackQueue first to make sure messages happen in order [self _performDelegateBlock:^{ dispatch_async(_workQueue, ^{ [self _sendFrameWithOpcode:SROpCodePong data:pingData]; }); }]; } - (void)handlePong; { // NOOP } - (void)_handleMessage:(id)message { SRFastLog(@"Received message"); [self _performDelegateBlock:^{ if ([self.delegate respondsToSelector:@selector(webSocket:didReceiveMessage:)]) { [self.delegate webSocket:self didReceiveMessage:message]; } }]; } static inline BOOL closeCodeIsValid(int closeCode) { if (closeCode < 1000) { return NO; } if (closeCode >= 1000 && closeCode <= 1011) { if (closeCode == 1004 || closeCode == 1005 || closeCode == 1006) { return NO; } return YES; } if (closeCode >= 3000 && closeCode <= 3999) { return YES; } if (closeCode >= 4000 && closeCode <= 4999) { return YES; } return NO; } // Note from RFC: // // If there is a body, the first two // bytes of the body MUST be a 2-byte unsigned integer (in network byte // order) representing a status code with value /code/ defined in // Section 7.4. Following the 2-byte integer the body MAY contain UTF-8 // encoded data with value /reason/, the interpretation of which is not // defined by this specification. - (void)handleCloseWithData:(NSData *)data; { size_t dataSize = data.length; __block uint16_t closeCode = 0; SRFastLog(@"Received close frame"); if (dataSize == 1) { // TODO handle error [self _closeWithProtocolError:@"Payload for close must be larger than 2 bytes"]; return; } else if (dataSize >= 2) { [data getBytes:&closeCode length:sizeof(closeCode)]; _closeCode = EndianU16_BtoN(closeCode); if (!closeCodeIsValid(_closeCode)) { [self _closeWithProtocolError:[NSString stringWithFormat:@"Cannot have close code of %d", _closeCode]]; return; } if (dataSize > 2) { _closeReason = [[NSString alloc] initWithData:[data subdataWithRange:NSMakeRange(2, dataSize - 2)] encoding:NSUTF8StringEncoding]; if (!_closeReason) { [self _closeWithProtocolError:@"Close reason MUST be valid UTF-8"]; return; } } } else { _closeCode = SRStatusNoStatusReceived; } [self assertOnWorkQueue]; if (self.readyState == SR_OPEN) { [self closeWithCode:1000 reason:nil]; } dispatch_async(_workQueue, ^{ [self _disconnect]; }); } - (void)_disconnect; { [self assertOnWorkQueue]; SRFastLog(@"Trying to disconnect"); _closeWhenFinishedWriting = YES; [self _pumpWriting]; } - (void)_handleFrameWithData:(NSData *)frameData opCode:(NSInteger)opcode; { // Check that the current data is valid UTF8 BOOL isControlFrame = (opcode == SROpCodePing || opcode == SROpCodePong || opcode == SROpCodeConnectionClose); if (!isControlFrame) { [self _readFrameNew]; } else { dispatch_async(_workQueue, ^{ [self _readFrameContinue]; }); } switch (opcode) { case SROpCodeTextFrame: { NSString *str = [[NSString alloc] initWithData:frameData encoding:NSUTF8StringEncoding]; if (str == nil && frameData) { [self closeWithCode:SRStatusCodeInvalidUTF8 reason:@"Text frames must be valid UTF-8"]; dispatch_async(_workQueue, ^{ [self _disconnect]; }); return; } [self _handleMessage:str]; break; } case SROpCodeBinaryFrame: [self _handleMessage:[frameData copy]]; break; case SROpCodeConnectionClose: [self handleCloseWithData:frameData]; break; case SROpCodePing: [self handlePing:frameData]; break; case SROpCodePong: [self handlePong]; break; default: [self _closeWithProtocolError:[NSString stringWithFormat:@"Unknown opcode %u", (int)opcode]]; // TODO: Handle invalid opcode break; } } - (void)_handleFrameHeader:(frame_header)frame_header curData:(NSData *)curData; { assert(frame_header.opcode != 0); if (self.readyState != SR_OPEN) { return; } BOOL isControlFrame = (frame_header.opcode == SROpCodePing || frame_header.opcode == SROpCodePong || frame_header.opcode == SROpCodeConnectionClose); if (isControlFrame && !frame_header.fin) { [self _closeWithProtocolError:@"Fragmented control frames not allowed"]; return; } if (isControlFrame && frame_header.payload_length >= 126) { [self _closeWithProtocolError:@"Control frames cannot have payloads larger than 126 bytes"]; return; } if (!isControlFrame) { _currentFrameOpcode = frame_header.opcode; _currentFrameCount += 1; } if (frame_header.payload_length == 0) { if (isControlFrame) { [self _handleFrameWithData:curData opCode:frame_header.opcode]; } else { if (frame_header.fin) { [self _handleFrameWithData:_currentFrameData opCode:frame_header.opcode]; } else { // TODO add assert that opcode is not a control; [self _readFrameContinue]; } } } else { [self _addConsumerWithDataLength:(size_t)frame_header.payload_length callback:^(FSRWebSocket *self, NSData *newData) { if (isControlFrame) { [self _handleFrameWithData:newData opCode:frame_header.opcode]; } else { if (frame_header.fin) { [self _handleFrameWithData:self->_currentFrameData opCode:frame_header.opcode]; } else { // TODO add assert that opcode is not a control; [self _readFrameContinue]; } } } readToCurrentFrame:!isControlFrame unmaskBytes:frame_header.masked]; } } /* From RFC: 0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +-+-+-+-+-------+-+-------------+-------------------------------+ |F|R|R|R| opcode|M| Payload len | Extended payload length | |I|S|S|S| (4) |A| (7) | (16/64) | |N|V|V|V| |S| | (if payload len==126/127) | | |1|2|3| |K| | | +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - + | Extended payload length continued, if payload len == 127 | + - - - - - - - - - - - - - - - +-------------------------------+ | |Masking-key, if MASK set to 1 | +-------------------------------+-------------------------------+ | Masking-key (continued) | Payload Data | +-------------------------------- - - - - - - - - - - - - - - - + : Payload Data continued ... : + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + | Payload Data continued ... | +---------------------------------------------------------------+ */ static const uint8_t SRFinMask = 0x80; static const uint8_t SROpCodeMask = 0x0F; static const uint8_t SRRsvMask = 0x70; static const uint8_t SRMaskMask = 0x80; static const uint8_t SRPayloadLenMask = 0x7F; - (void)_readFrameContinue; { assert((_currentFrameCount == 0 && _currentFrameOpcode == 0) || (_currentFrameCount > 0 && _currentFrameOpcode > 0)); [self _addConsumerWithDataLength:2 callback:^(FSRWebSocket *self, NSData *data) { __block frame_header header = {0}; const uint8_t *headerBuffer = data.bytes; assert(data.length >= 2); if (headerBuffer[0] & SRRsvMask) { [self _closeWithProtocolError:@"Server used RSV bits"]; return; } uint8_t receivedOpcode = (SROpCodeMask & headerBuffer[0]); BOOL isControlFrame = (receivedOpcode == SROpCodePing || receivedOpcode == SROpCodePong || receivedOpcode == SROpCodeConnectionClose); if (!isControlFrame && receivedOpcode != 0 && self->_currentFrameCount > 0) { [self _closeWithProtocolError:@"all data frames after the initial data frame must have opcode 0"]; return; } if (receivedOpcode == 0 && self->_currentFrameCount == 0) { [self _closeWithProtocolError:@"cannot continue a message"]; return; } header.opcode = receivedOpcode == 0 ? self->_currentFrameOpcode : receivedOpcode; header.fin = !!(SRFinMask & headerBuffer[0]); header.masked = !!(SRMaskMask & headerBuffer[1]); header.payload_length = SRPayloadLenMask & headerBuffer[1]; headerBuffer = NULL; if (header.masked) { [self _closeWithProtocolError:@"Client must receive unmasked data"]; } size_t extra_bytes_needed = header.masked ? sizeof(_currentReadMaskKey) : 0; if (header.payload_length == 126) { extra_bytes_needed += sizeof(uint16_t); } else if (header.payload_length == 127) { extra_bytes_needed += sizeof(uint64_t); } if (extra_bytes_needed == 0) { [self _handleFrameHeader:header curData:self->_currentFrameData]; } else { [self _addConsumerWithDataLength:extra_bytes_needed callback:^(FSRWebSocket *self, NSData *data) { size_t mapped_size = data.length; const void *mapped_buffer = data.bytes; size_t offset = 0; if (header.payload_length == 126) { assert(mapped_size >= sizeof(uint16_t)); uint16_t newLen = EndianU16_BtoN(*(uint16_t *)(mapped_buffer)); header.payload_length = newLen; offset += sizeof(uint16_t); } else if (header.payload_length == 127) { assert(mapped_size >= sizeof(uint64_t)); header.payload_length = EndianU64_BtoN(*(uint64_t *)(mapped_buffer)); offset += sizeof(uint64_t); } else { assert(header.payload_length < 126 && header.payload_length >= 0); } if (header.masked) { assert(mapped_size >= sizeof(_currentReadMaskOffset) + offset); memcpy(self->_currentReadMaskKey, ((uint8_t *)mapped_buffer) + offset, sizeof(self->_currentReadMaskKey)); } [self _handleFrameHeader:header curData:self->_currentFrameData]; } readToCurrentFrame:NO unmaskBytes:NO]; } } readToCurrentFrame:NO unmaskBytes:NO]; } - (void)_readFrameNew; { dispatch_async(_workQueue, ^{ [_currentFrameData setLength:0]; _currentFrameOpcode = 0; _currentFrameCount = 0; _readOpCount = 0; _currentStringScanPosition = 0; [self _readFrameContinue]; }); } - (void)_pumpWriting; { [self assertOnWorkQueue]; NSUInteger dataLength = _outputBuffer.length; if (dataLength - _outputBufferOffset > 0 && _outputStream.hasSpaceAvailable) { NSUInteger bytesWritten = [_outputStream write:_outputBuffer.bytes + _outputBufferOffset maxLength:dataLength - _outputBufferOffset]; if (bytesWritten == -1) { [self _failWithError:[NSError errorWithDomain:@"org.lolrus.SocketRocket" code:2145 userInfo:[NSDictionary dictionaryWithObject:@"Error writing to stream" forKey:NSLocalizedDescriptionKey]]]; return; } _outputBufferOffset += bytesWritten; if (_outputBufferOffset > 4096 && _outputBufferOffset > (_outputBuffer.length >> 1)) { _outputBuffer = [[NSMutableData alloc] initWithBytes:(char *)_outputBuffer.bytes + _outputBufferOffset length:_outputBuffer.length - _outputBufferOffset]; _outputBufferOffset = 0; } } if (_closeWhenFinishedWriting && _outputBuffer.length - _outputBufferOffset == 0 && (_inputStream.streamStatus != NSStreamStatusNotOpen && _inputStream.streamStatus != NSStreamStatusClosed) && !_sentClose) { _sentClose = YES; @synchronized (self) { [_outputStream close]; [_inputStream close]; // TODO: Why are we missing the SocketRocket code to call unscheduleFromRunLoop??? } if (!_failed) { [self _performDelegateBlock:^{ if ([self.delegate respondsToSelector:@selector(webSocket:didCloseWithCode:reason:wasClean:)]) { [self.delegate webSocket:self didCloseWithCode:_closeCode reason:_closeReason wasClean:YES]; } }]; } [self _scheduleCleanup]; } } - (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback; { [self assertOnWorkQueue]; [self _addConsumerWithScanner:consumer callback:callback dataLength:0]; } - (void)_addConsumerWithDataLength:(size_t)dataLength callback:(data_callback)callback readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes; { [self assertOnWorkQueue]; assert(dataLength); [_consumers addObject:[_consumerPool consumerWithScanner:nil handler:callback bytesNeeded:dataLength readToCurrentFrame:readToCurrentFrame unmaskBytes:unmaskBytes]]; [self _pumpScanner]; } - (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback dataLength:(size_t)dataLength; { [self assertOnWorkQueue]; [_consumers addObject:[_consumerPool consumerWithScanner:consumer handler:callback bytesNeeded:dataLength readToCurrentFrame:NO unmaskBytes:NO]]; [self _pumpScanner]; } - (void)_scheduleCleanup { @synchronized(self) { if (_cleanupScheduled) { return; } _cleanupScheduled = YES; // Cleanup NSStream delegate's in the same RunLoop used by the streams themselves: // This way we'll prevent race conditions between handleEvent and SRWebsocket's dealloc NSTimer *timer = [NSTimer timerWithTimeInterval:(0.0f) target:self selector:@selector(_cleanupSelfReference:) userInfo:nil repeats:NO]; [[NSRunLoop FSR_networkRunLoop] addTimer:timer forMode:NSDefaultRunLoopMode]; } } - (void)_cleanupSelfReference:(NSTimer *)timer { @synchronized(self) { // Nuke NSStream delegate's _inputStream.delegate = nil; _outputStream.delegate = nil; // Remove the streams, right now, from the networkRunLoop [_inputStream close]; [_outputStream close]; } // Cleanup selfRetain in the same GCD queue as usual dispatch_async(_workQueue, ^{ _selfRetain = nil; }); } static const char CRLFCRLFBytes[] = {'\r', '\n', '\r', '\n'}; - (void)_readUntilHeaderCompleteWithCallback:(data_callback)dataHandler; { [self _readUntilBytes:CRLFCRLFBytes length:sizeof(CRLFCRLFBytes) callback:dataHandler]; } - (void)_readUntilBytes:(const void *)bytes length:(size_t)length callback:(data_callback)dataHandler; { // TODO optimize so this can continue from where we last searched stream_scanner consumer = ^size_t(NSData *data) { __block size_t found_size = 0; __block size_t match_count = 0; size_t size = data.length; const unsigned char *buffer = data.bytes; for (int i = 0; i < size; i++ ) { if (((const unsigned char *)buffer)[i] == ((const unsigned char *)bytes)[match_count]) { match_count += 1; if (match_count == length) { found_size = i + 1; break; } } else { match_count = 0; } } return found_size; }; [self _addConsumerWithScanner:consumer callback:dataHandler]; } // Returns true if did work - (BOOL)_innerPumpScanner { BOOL didWork = NO; if (self.readyState >= SR_CLOSING) { return didWork; } if (!_consumers.count) { return didWork; } size_t curSize = _readBuffer.length - _readBufferOffset; if (!curSize) { return didWork; } FSRIOConsumer *consumer = [_consumers objectAtIndex:0]; size_t bytesNeeded = consumer.bytesNeeded; size_t foundSize = 0; if (consumer.consumer) { NSData *tempView = [NSData dataWithBytesNoCopy:(char *)_readBuffer.bytes + _readBufferOffset length:_readBuffer.length - _readBufferOffset freeWhenDone:NO]; foundSize = consumer.consumer(tempView); } else { assert(consumer.bytesNeeded); if (curSize >= bytesNeeded) { foundSize = bytesNeeded; } else if (consumer.readToCurrentFrame) { foundSize = curSize; } } NSData *slice = nil; if (consumer.readToCurrentFrame || foundSize) { NSRange sliceRange = NSMakeRange(_readBufferOffset, foundSize); slice = [_readBuffer subdataWithRange:sliceRange]; _readBufferOffset += foundSize; if (_readBufferOffset > 4096 && _readBufferOffset > (_readBuffer.length >> 1)) { _readBuffer = [[NSMutableData alloc] initWithBytes:(char *)_readBuffer.bytes + _readBufferOffset length:_readBuffer.length - _readBufferOffset]; _readBufferOffset = 0; } if (consumer.unmaskBytes) { NSMutableData *mutableSlice = [slice mutableCopy]; NSUInteger len = mutableSlice.length; uint8_t *bytes = mutableSlice.mutableBytes; for (int i = 0; i < len; i++) { bytes[i] = bytes[i] ^ _currentReadMaskKey[_currentReadMaskOffset % sizeof(_currentReadMaskKey)]; _currentReadMaskOffset += 1; } slice = mutableSlice; } if (consumer.readToCurrentFrame) { [_currentFrameData appendData:slice]; _readOpCount += 1; if (_currentFrameOpcode == SROpCodeTextFrame) { // Validate UTF8 stuff. size_t currentDataSize = _currentFrameData.length; if (_currentFrameOpcode == SROpCodeTextFrame && currentDataSize > 0) { // TODO: Optimize the crap out of this. Don't really have to copy all the data each time size_t scanSize = currentDataSize - _currentStringScanPosition; NSData *scan_data = [_currentFrameData subdataWithRange:NSMakeRange(_currentStringScanPosition, scanSize)]; int32_t valid_utf8_size = validate_dispatch_data_partial_string(scan_data); if (valid_utf8_size == -1) { [self closeWithCode:SRStatusCodeInvalidUTF8 reason:@"Text frames must be valid UTF-8"]; dispatch_async(_workQueue, ^{ [self _disconnect]; }); return didWork; } else { _currentStringScanPosition += valid_utf8_size; } } } consumer.bytesNeeded -= foundSize; if (consumer.bytesNeeded == 0) { [_consumers removeObjectAtIndex:0]; consumer.handler(self, nil); didWork = YES; } } else if (foundSize) { [_consumers removeObjectAtIndex:0]; consumer.handler(self, slice); didWork = YES; } } return didWork; } -(void)_pumpScanner; { [self assertOnWorkQueue]; if (!_isPumping) { _isPumping = YES; } else { return; } while ([self _innerPumpScanner]) { } _isPumping = NO; } //#define NOMASK static const size_t SRFrameHeaderOverhead = 32; - (void)_sendFrameWithOpcode:(FSROpCode)opcode data:(id)data; { [self assertOnWorkQueue]; NSAssert(data == nil || [data isKindOfClass:[NSData class]] || [data isKindOfClass:[NSString class]], @"Function expects nil, NSString or NSData"); size_t payloadLength = [data isKindOfClass:[NSString class]] ? [(NSString *)data lengthOfBytesUsingEncoding:NSUTF8StringEncoding] : [data length]; NSMutableData *frame = [[NSMutableData alloc] initWithLength:payloadLength + SRFrameHeaderOverhead]; if (!frame) { [self closeWithCode:SRStatusCodeMessageTooBig reason:@"Message too big"]; return; } uint8_t *frame_buffer = (uint8_t *)[frame mutableBytes]; // set fin frame_buffer[0] = SRFinMask | opcode; BOOL useMask = YES; #ifdef NOMASK useMask = NO; #endif if (useMask) { // set the mask and header frame_buffer[1] |= SRMaskMask; } size_t frame_buffer_size = 2; const uint8_t *unmasked_payload = NULL; if ([data isKindOfClass:[NSData class]]) { unmasked_payload = (uint8_t *)[data bytes]; } else if ([data isKindOfClass:[NSString class]]) { unmasked_payload = (const uint8_t *)[data UTF8String]; } else { assert(NO); } if (payloadLength < 126) { frame_buffer[1] |= payloadLength; } else if (payloadLength <= UINT16_MAX) { frame_buffer[1] |= 126; *((uint16_t *)(frame_buffer + frame_buffer_size)) = EndianU16_BtoN((uint16_t)payloadLength); frame_buffer_size += sizeof(uint16_t); } else { frame_buffer[1] |= 127; *((uint64_t *)(frame_buffer + frame_buffer_size)) = EndianU64_BtoN((uint64_t)payloadLength); frame_buffer_size += sizeof(uint64_t); } if (!useMask) { for (int i = 0; i < payloadLength; i++) { frame_buffer[frame_buffer_size] = unmasked_payload[i]; frame_buffer_size += 1; } } else { uint8_t *mask_key = frame_buffer + frame_buffer_size; int result = SecRandomCopyBytes(kSecRandomDefault, sizeof(uint32_t), (uint8_t *)mask_key); assert(result == 0); frame_buffer_size += sizeof(uint32_t); // TODO: could probably optimize this with SIMD for (int i = 0; i < payloadLength; i++) { frame_buffer[frame_buffer_size] = unmasked_payload[i] ^ mask_key[i % sizeof(uint32_t)]; frame_buffer_size += 1; } } assert(frame_buffer_size <= [frame length]); frame.length = frame_buffer_size; [self _writeData:frame]; } - (void)stream:(NSStream *)aStream handleEvent:(NSStreamEvent)eventCode; { __weak __typeof__(self) weakSelf = self; // turn on keep-alive for the output stream. if (eventCode == NSStreamEventOpenCompleted && aStream == _outputStream) { CFDataRef socketData = CFWriteStreamCopyProperty((CFWriteStreamRef)_outputStream, kCFStreamPropertySocketNativeHandle); // In rare cases socketData might be nil (there are crash reports out there), in which case we'll have to just // live without keep-alive :( if (socketData != nil) { CFSocketNativeHandle socket; CFDataGetBytes(socketData, CFRangeMake(0, sizeof(CFSocketNativeHandle)), (UInt8 *)&socket); CFRelease(socketData); int keepAliveOn = 1; if (setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, &keepAliveOn, sizeof(keepAliveOn)) == -1) { SRFastLog(@"Failed to turn on TCP keepalive for websocket"); } } } if (_secure && !_pinnedCertFound && (eventCode == NSStreamEventHasBytesAvailable || eventCode == NSStreamEventHasSpaceAvailable)) { NSArray *sslCerts = [_urlRequest FSR_SSLPinnedCertificates]; if (sslCerts) { SecTrustRef secTrust = (__bridge SecTrustRef)[aStream propertyForKey:(__bridge id)kCFStreamPropertySSLPeerTrust]; if (secTrust) { NSInteger numCerts = SecTrustGetCertificateCount(secTrust); for (NSInteger i = 0; i < numCerts && !_pinnedCertFound; i++) { SecCertificateRef cert = SecTrustGetCertificateAtIndex(secTrust, i); NSData *certData = CFBridgingRelease(SecCertificateCopyData(cert)); for (id ref in sslCerts) { SecCertificateRef trustedCert = (__bridge SecCertificateRef)ref; NSData *trustedCertData = CFBridgingRelease(SecCertificateCopyData(trustedCert)); if ([trustedCertData isEqualToData:certData]) { _pinnedCertFound = YES; break; } } } } if (!_pinnedCertFound) { dispatch_async(_workQueue, ^{ NSDictionary *userInfo = @{ NSLocalizedDescriptionKey : @"Invalid server cert" }; [weakSelf _failWithError:[NSError errorWithDomain:@"org.lolrus.SocketRocket" code:23556 userInfo:userInfo]]; }); return; } } } // SRFastLog(@"%@ Got stream event %d", aStream, eventCode); dispatch_async(_workQueue, ^{ [weakSelf safeHandleEvent:eventCode stream:aStream]; }); } - (void)safeHandleEvent:(NSStreamEvent)eventCode stream:(NSStream *)aStream { switch (eventCode) { case NSStreamEventOpenCompleted: { SRFastLog(@"NSStreamEventOpenCompleted %@", aStream); if (self.readyState >= SR_CLOSING) { return; } assert(_readBuffer); if (self.readyState == SR_CONNECTING && aStream == _inputStream) { [self didConnect]; } [self _pumpWriting]; [self _pumpScanner]; break; } case NSStreamEventErrorOccurred: { SRFastLog(@"NSStreamEventErrorOccurred %@ %@", aStream, [[aStream streamError] copy]); /// TODO specify error better! [self _failWithError:aStream.streamError]; _readBufferOffset = 0; [_readBuffer setLength:0]; break; } case NSStreamEventEndEncountered: { [self _pumpScanner]; SRFastLog(@"NSStreamEventEndEncountered %@", aStream); if (aStream.streamError) { [self _failWithError:aStream.streamError]; } else { dispatch_async(_workQueue, ^{ if (self.readyState != SR_CLOSED) { self.readyState = SR_CLOSED; [self _scheduleCleanup]; } if (!_sentClose && !_failed) { _sentClose = YES; // If we get closed in this state it's probably not clean because we should be sending this when we send messages [self _performDelegateBlock:^{ if ([self.delegate respondsToSelector:@selector(webSocket:didCloseWithCode:reason:wasClean:)]) { [self.delegate webSocket:self didCloseWithCode:0 reason:@"Stream end encountered" wasClean:NO]; } }]; } }); } break; } case NSStreamEventHasBytesAvailable: { SRFastLog(@"NSStreamEventHasBytesAvailable %@", aStream); const NSUInteger bufferSize = 2048; uint8_t buffer[bufferSize]; while (_inputStream.hasBytesAvailable) { NSInteger bytes_read = [_inputStream read:buffer maxLength:bufferSize]; if (bytes_read > 0) { [_readBuffer appendBytes:buffer length:bytes_read]; } else if (bytes_read < 0) { [self _failWithError:_inputStream.streamError]; } if (bytes_read != bufferSize) { break; } }; [self _pumpScanner]; break; } case NSStreamEventHasSpaceAvailable: { SRFastLog(@"NSStreamEventHasSpaceAvailable %@", aStream); [self _pumpWriting]; break; } default: SRFastLog(@"(default) %@", aStream); break; } } @end @implementation FSRIOConsumer @synthesize bytesNeeded = _bytesNeeded; @synthesize consumer = _scanner; @synthesize handler = _handler; @synthesize readToCurrentFrame = _readToCurrentFrame; @synthesize unmaskBytes = _unmaskBytes; - (void)setupWithScanner:(stream_scanner)scanner handler:(data_callback)handler bytesNeeded:(size_t)bytesNeeded readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes; { _scanner = [scanner copy]; _handler = [handler copy]; _bytesNeeded = bytesNeeded; _readToCurrentFrame = readToCurrentFrame; _unmaskBytes = unmaskBytes; assert(_scanner || _bytesNeeded); } @end @implementation FSRIOConsumerPool { NSUInteger _poolSize; NSMutableArray *_bufferedConsumers; } - (id)initWithBufferCapacity:(NSUInteger)poolSize; { self = [super init]; if (self) { _poolSize = poolSize; _bufferedConsumers = [[NSMutableArray alloc] initWithCapacity:poolSize]; } return self; } - (id)init { return [self initWithBufferCapacity:8]; } - (FSRIOConsumer *)consumerWithScanner:(stream_scanner)scanner handler:(data_callback)handler bytesNeeded:(size_t)bytesNeeded readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes; { FSRIOConsumer *consumer = nil; if (_bufferedConsumers.count) { consumer = [_bufferedConsumers lastObject]; [_bufferedConsumers removeLastObject]; } else { consumer = [[FSRIOConsumer alloc] init]; } [consumer setupWithScanner:scanner handler:handler bytesNeeded:bytesNeeded readToCurrentFrame:readToCurrentFrame unmaskBytes:unmaskBytes]; return consumer; } - (void)returnConsumer:(FSRIOConsumer *)consumer; { if (_bufferedConsumers.count < _poolSize) { [_bufferedConsumers addObject:consumer]; } } @end @implementation NSURLRequest (FCertificateAdditions) - (NSArray *)FSR_SSLPinnedCertificates; { return [NSURLProtocol propertyForKey:@"FSR_SSLPinnedCertificates" inRequest:self]; } @end @implementation NSMutableURLRequest (FCertificateAdditions) - (NSArray *)FSR_SSLPinnedCertificates; { return [NSURLProtocol propertyForKey:@"FSR_SSLPinnedCertificates" inRequest:self]; } - (void)setFSR_SSLPinnedCertificates:(NSArray *)FSR_SSLPinnedCertificates; { [NSURLProtocol setProperty:FSR_SSLPinnedCertificates forKey:@"FSR_SSLPinnedCertificates" inRequest:self]; } @end @implementation NSURL (FSRWebSocket) - (NSString *)SR_origin; { NSString *scheme = [self.scheme lowercaseString]; if ([scheme isEqualToString:@"wss"]) { scheme = @"https"; } else if ([scheme isEqualToString:@"ws"]) { scheme = @"http"; } if (self.port) { return [NSString stringWithFormat:@"%@://%@:%@/", scheme, self.host, self.port]; } else { return [NSString stringWithFormat:@"%@://%@/", scheme, self.host]; } } @end // #define SR_ENABLE_LOG static inline void SRFastLog(NSString *format, ...) { #ifdef SR_ENABLE_LOG __block va_list arg_list; va_start (arg_list, format); NSString *formattedString = [[NSString alloc] initWithFormat:format arguments:arg_list]; va_end(arg_list); NSLog(@"[SR] %@", formattedString); #endif } #ifdef HAS_ICU static inline int32_t validate_dispatch_data_partial_string(NSData *data) { const void * contents = [data bytes]; long size = [data length]; const uint8_t *str = (const uint8_t *)contents; UChar32 codepoint = 1; int32_t offset = 0; int32_t lastOffset = 0; while(offset < size && codepoint > 0) { lastOffset = offset; U8_NEXT(str, offset, size, codepoint); } if (codepoint == -1) { // Check to see if the last byte is valid or whether it was just continuing if (!U8_IS_LEAD(str[lastOffset]) || U8_COUNT_TRAIL_BYTES(str[lastOffset]) + lastOffset < (int32_t)size) { size = -1; } else { uint8_t leadByte = str[lastOffset]; U8_MASK_LEAD_BYTE(leadByte, U8_COUNT_TRAIL_BYTES(leadByte)); for (int i = lastOffset + 1; i < offset; i++) { if (U8_IS_SINGLE(str[i]) || U8_IS_LEAD(str[i]) || !U8_IS_TRAIL(str[i])) { size = -1; } } if (size != -1) { size = lastOffset; } } } if (size != -1 && ![[NSString alloc] initWithBytesNoCopy:(char *)[data bytes] length:size encoding:NSUTF8StringEncoding freeWhenDone:NO]) { size = -1; } return (int32_t)size; } #else // This is a hack, and probably not optimal static inline int32_t validate_dispatch_data_partial_string(NSData *data) { static const int maxCodepointSize = 3; for (int i = 0; i < maxCodepointSize; i++) { NSString *str = [[NSString alloc] initWithBytesNoCopy:(char *)data.bytes length:data.length - i encoding:NSUTF8StringEncoding freeWhenDone:NO]; if (str) { return (int)(data.length - i); } } return -1; } #endif static _FSRRunLoopThread *networkThread = nil; static NSRunLoop *networkRunLoop = nil; @implementation NSRunLoop (FSRWebSocket) + (NSRunLoop *)FSR_networkRunLoop { static dispatch_once_t onceToken; dispatch_once(&onceToken, ^{ networkThread = [[_FSRRunLoopThread alloc] init]; networkThread.name = @"com.squareup.SocketRocket.NetworkThread"; [networkThread start]; networkRunLoop = networkThread.runLoop; }); return networkRunLoop; } @end @implementation _FSRRunLoopThread { dispatch_group_t _waitGroup; } @synthesize runLoop = _runLoop; - (void)dealloc { sr_dispatch_release(_waitGroup); } - (id)init { self = [super init]; if (self) { _waitGroup = dispatch_group_create(); dispatch_group_enter(_waitGroup); } return self; } /** * This is the main method of the thread on which the socket events are scheduled in a run loop. */ - (void)main; { @autoreleasepool { _runLoop = [NSRunLoop currentRunLoop]; dispatch_group_leave(_waitGroup); // Add an empty run loop source to prevent runloop from spinning. CFRunLoopSourceContext sourceCtx = { .version = 0, .info = NULL, .retain = NULL, .release = NULL, .copyDescription = NULL, .equal = NULL, .hash = NULL, .schedule = NULL, .cancel = NULL, .perform = NULL }; CFRunLoopSourceRef source = CFRunLoopSourceCreate(NULL, 0, &sourceCtx); CFRunLoopAddSource(CFRunLoopGetCurrent(), source, kCFRunLoopDefaultMode); CFRelease(source); while ([_runLoop runMode:NSDefaultRunLoopMode beforeDate:[NSDate distantFuture]]) { } assert(NO); } } - (NSRunLoop *)runLoop; { dispatch_group_wait(_waitGroup, DISPATCH_TIME_FOREVER); return _runLoop; } @end