1//
2//   Copyright 2012 Square Inc.
3//
4//   Licensed under the Apache License, Version 2.0 (the "License");
5//   you may not use this file except in compliance with the License.
6//   You may obtain a copy of the License at
7//
8//       http://www.apache.org/licenses/LICENSE-2.0
9//
10//   Unless required by applicable law or agreed to in writing, software
11//   distributed under the License is distributed on an "AS IS" BASIS,
12//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13//   See the License for the specific language governing permissions and
14//   limitations under the License.
15//
16
17
18#import "SRWebSocket.h"
19
20#if TARGET_OS_IPHONE
21#define HAS_ICU
22#endif
23
24#ifdef HAS_ICU
25#import <unicode/utf8.h>
26#endif
27
28#if TARGET_OS_IPHONE
29#import <Endian.h>
30#else
31#import <CoreServices/CoreServices.h>
32#endif
33
34#import <CommonCrypto/CommonDigest.h>
35#import <Security/SecRandom.h>
36
37#if OS_OBJECT_USE_OBJC_RETAIN_RELEASE
38#define sr_dispatch_retain(x)
39#define sr_dispatch_release(x)
40#define maybe_bridge(x) ((__bridge void *) x)
41#else
42#define sr_dispatch_retain(x) dispatch_retain(x)
43#define sr_dispatch_release(x) dispatch_release(x)
44#define maybe_bridge(x) (x)
45#endif
46
47#if !__has_feature(objc_arc)
48#error SocketRocket must be compiled with ARC enabled
49#endif
50
51
52typedef enum  {
53    SROpCodeTextFrame = 0x1,
54    SROpCodeBinaryFrame = 0x2,
55    // 3-7 reserved.
56    SROpCodeConnectionClose = 0x8,
57    SROpCodePing = 0x9,
58    SROpCodePong = 0xA,
59    // B-F reserved.
60} SROpCode;
61
62typedef struct {
63    BOOL fin;
64//  BOOL rsv1;
65//  BOOL rsv2;
66//  BOOL rsv3;
67    uint8_t opcode;
68    BOOL masked;
69    uint64_t payload_length;
70} frame_header;
71
72static NSString *const SRWebSocketAppendToSecKeyString = @"258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
73
74static inline int32_t validate_dispatch_data_partial_string(NSData *data);
75static inline void SRFastLog(NSString *format, ...);
76
77@interface NSData (SRWebSocket)
78
79- (NSString *)stringBySHA1ThenBase64Encoding;
80
81@end
82
83
84@interface NSString (SRWebSocket)
85
86- (NSString *)stringBySHA1ThenBase64Encoding;
87
88@end
89
90
91@interface NSURL (SRWebSocket)
92
93// The origin isn't really applicable for a native application.
94// So instead, just map ws -> http and wss -> https.
95- (NSString *)SR_origin;
96
97@end
98
99
100@interface _SRRunLoopThread : NSThread
101
102@property (nonatomic, readonly) NSRunLoop *runLoop;
103
104@end
105
106
107static NSString *newSHA1String(const char *bytes, size_t length) {
108    uint8_t md[CC_SHA1_DIGEST_LENGTH];
109
110    assert(length >= 0);
111    assert(length <= UINT32_MAX);
112    CC_SHA1(bytes, (CC_LONG)length, md);
113
114    NSData *data = [NSData dataWithBytes:md length:CC_SHA1_DIGEST_LENGTH];
115
116    if ([data respondsToSelector:@selector(base64EncodedStringWithOptions:)]) {
117        return [data base64EncodedStringWithOptions:0];
118    }
119
120    return [data base64Encoding];
121}
122
123@implementation NSData (SRWebSocket)
124
125- (NSString *)stringBySHA1ThenBase64Encoding;
126{
127    return newSHA1String(self.bytes, self.length);
128}
129
130@end
131
132
133@implementation NSString (SRWebSocket)
134
135- (NSString *)stringBySHA1ThenBase64Encoding;
136{
137    return newSHA1String(self.UTF8String, self.length);
138}
139
140@end
141
142NSString *const SRWebSocketErrorDomain = @"SRWebSocketErrorDomain";
143NSString *const SRHTTPResponseErrorKey = @"HTTPResponseStatusCode";
144
145// Returns number of bytes consumed. Returning 0 means you didn't match.
146// Sends bytes to callback handler;
147typedef size_t (^stream_scanner)(NSData *collected_data);
148
149typedef void (^data_callback)(SRWebSocket *webSocket,  NSData *data);
150
151@interface SRIOConsumer : NSObject {
152    stream_scanner _scanner;
153    data_callback _handler;
154    size_t _bytesNeeded;
155    BOOL _readToCurrentFrame;
156    BOOL _unmaskBytes;
157}
158@property (nonatomic, copy, readonly) stream_scanner consumer;
159@property (nonatomic, copy, readonly) data_callback handler;
160@property (nonatomic, assign) size_t bytesNeeded;
161@property (nonatomic, assign, readonly) BOOL readToCurrentFrame;
162@property (nonatomic, assign, readonly) BOOL unmaskBytes;
163
164@end
165
166// This class is not thread-safe, and is expected to always be run on the same queue.
167@interface SRIOConsumerPool : NSObject
168
169- (id)initWithBufferCapacity:(NSUInteger)poolSize;
170
171- (SRIOConsumer *)consumerWithScanner:(stream_scanner)scanner handler:(data_callback)handler bytesNeeded:(size_t)bytesNeeded readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes;
172- (void)returnConsumer:(SRIOConsumer *)consumer;
173
174@end
175
176@interface SRWebSocket ()  <NSStreamDelegate>
177
178- (void)_writeData:(NSData *)data;
179- (void)_closeWithProtocolError:(NSString *)message;
180- (void)_failWithError:(NSError *)error;
181
182- (void)_disconnect;
183
184- (void)_readFrameNew;
185- (void)_readFrameContinue;
186
187- (void)_pumpScanner;
188
189- (void)_pumpWriting;
190
191- (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback;
192- (void)_addConsumerWithDataLength:(size_t)dataLength callback:(data_callback)callback readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes;
193- (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback dataLength:(size_t)dataLength;
194- (void)_readUntilBytes:(const void *)bytes length:(size_t)length callback:(data_callback)dataHandler;
195- (void)_readUntilHeaderCompleteWithCallback:(data_callback)dataHandler;
196
197- (void)_sendFrameWithOpcode:(SROpCode)opcode data:(id)data;
198
199- (BOOL)_checkHandshake:(CFHTTPMessageRef)httpMessage;
200- (void)_SR_commonInit;
201
202- (void)_initializeStreams;
203- (void)_connect;
204
205@property (nonatomic) SRReadyState readyState;
206
207@property (nonatomic) NSOperationQueue *delegateOperationQueue;
208@property (nonatomic) dispatch_queue_t delegateDispatchQueue;
209
210@end
211
212
213@implementation SRWebSocket {
214    NSInteger _webSocketVersion;
215
216    NSOperationQueue *_delegateOperationQueue;
217    dispatch_queue_t _delegateDispatchQueue;
218
219    dispatch_queue_t _workQueue;
220    NSMutableArray *_consumers;
221
222    NSInputStream *_inputStream;
223    NSOutputStream *_outputStream;
224
225    NSMutableData *_readBuffer;
226    NSUInteger _readBufferOffset;
227
228    NSMutableData *_outputBuffer;
229    NSUInteger _outputBufferOffset;
230
231    uint8_t _currentFrameOpcode;
232    size_t _currentFrameCount;
233    size_t _readOpCount;
234    uint32_t _currentStringScanPosition;
235    NSMutableData *_currentFrameData;
236
237    NSString *_closeReason;
238
239    NSString *_secKey;
240
241    BOOL _pinnedCertFound;
242
243    uint8_t _currentReadMaskKey[4];
244    size_t _currentReadMaskOffset;
245
246    BOOL _consumerStopped;
247
248    BOOL _closeWhenFinishedWriting;
249    BOOL _failed;
250
251    BOOL _secure;
252    NSURLRequest *_urlRequest;
253
254    CFHTTPMessageRef _receivedHTTPHeaders;
255
256    BOOL _sentClose;
257    BOOL _didFail;
258    int _closeCode;
259
260    BOOL _isPumping;
261
262    NSMutableSet *_scheduledRunloops;
263
264    // We use this to retain ourselves.
265    __strong SRWebSocket *_selfRetain;
266
267    NSArray *_requestedProtocols;
268    SRIOConsumerPool *_consumerPool;
269}
270
271@synthesize delegate = _delegate;
272@synthesize url = _url;
273@synthesize readyState = _readyState;
274@synthesize protocol = _protocol;
275
276static __strong NSData *CRLFCRLF;
277
278+ (void)initialize;
279{
280    CRLFCRLF = [[NSData alloc] initWithBytes:"\r\n\r\n" length:4];
281}
282
283- (id)initWithURLRequest:(NSURLRequest *)request protocols:(NSArray *)protocols;
284{
285    self = [super init];
286    if (self) {
287        assert(request.URL);
288        _url = request.URL;
289        _urlRequest = request;
290
291        _requestedProtocols = [protocols copy];
292
293        [self _SR_commonInit];
294    }
295
296    return self;
297}
298
299- (id)initWithURLRequest:(NSURLRequest *)request;
300{
301    return [self initWithURLRequest:request protocols:nil];
302}
303
304- (id)initWithURL:(NSURL *)url;
305{
306    return [self initWithURL:url protocols:nil];
307}
308
309- (id)initWithURL:(NSURL *)url protocols:(NSArray *)protocols;
310{
311    NSMutableURLRequest *request = [[NSMutableURLRequest alloc] initWithURL:url];
312    return [self initWithURLRequest:request protocols:protocols];
313}
314
315- (void)_SR_commonInit;
316{
317
318    NSString *scheme = _url.scheme.lowercaseString;
319    assert([scheme isEqualToString:@"ws"] || [scheme isEqualToString:@"http"] || [scheme isEqualToString:@"wss"] || [scheme isEqualToString:@"https"]);
320
321    if ([scheme isEqualToString:@"wss"] || [scheme isEqualToString:@"https"]) {
322        _secure = YES;
323    }
324
325    _readyState = SR_CONNECTING;
326    _consumerStopped = YES;
327    _webSocketVersion = 13;
328
329    _workQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL);
330
331    // Going to set a specific on the queue so we can validate we're on the work queue
332    dispatch_queue_set_specific(_workQueue, (__bridge void *)self, maybe_bridge(_workQueue), NULL);
333
334    _delegateDispatchQueue = dispatch_get_main_queue();
335    sr_dispatch_retain(_delegateDispatchQueue);
336
337    _readBuffer = [[NSMutableData alloc] init];
338    _outputBuffer = [[NSMutableData alloc] init];
339
340    _currentFrameData = [[NSMutableData alloc] init];
341
342    _consumers = [[NSMutableArray alloc] init];
343
344    _consumerPool = [[SRIOConsumerPool alloc] init];
345
346    _scheduledRunloops = [[NSMutableSet alloc] init];
347
348    [self _initializeStreams];
349
350    // default handlers
351}
352
353- (void)assertOnWorkQueue;
354{
355    assert(dispatch_get_specific((__bridge void *)self) == maybe_bridge(_workQueue));
356}
357
358- (void)dealloc
359{
360    _inputStream.delegate = nil;
361    _outputStream.delegate = nil;
362
363    [_inputStream close];
364    [_outputStream close];
365
366    sr_dispatch_release(_workQueue);
367    _workQueue = NULL;
368
369    if (_receivedHTTPHeaders) {
370        CFRelease(_receivedHTTPHeaders);
371        _receivedHTTPHeaders = NULL;
372    }
373
374    if (_delegateDispatchQueue) {
375        sr_dispatch_release(_delegateDispatchQueue);
376        _delegateDispatchQueue = NULL;
377    }
378}
379
380#ifndef NDEBUG
381
382- (void)setReadyState:(SRReadyState)aReadyState;
383{
384    [self willChangeValueForKey:@"readyState"];
385    assert(aReadyState > _readyState);
386    _readyState = aReadyState;
387    [self didChangeValueForKey:@"readyState"];
388}
389
390#endif
391
392- (void)open;
393{
394    assert(_url);
395    NSAssert(_readyState == SR_CONNECTING, @"Cannot call -(void)open on SRWebSocket more than once");
396
397    _selfRetain = self;
398
399    [self _connect];
400}
401
402// Calls block on delegate queue
403- (void)_performDelegateBlock:(dispatch_block_t)block;
404{
405    if (_delegateOperationQueue) {
406        [_delegateOperationQueue addOperationWithBlock:block];
407    } else {
408        assert(_delegateDispatchQueue);
409        dispatch_async(_delegateDispatchQueue, block);
410    }
411}
412
413- (void)setDelegateDispatchQueue:(dispatch_queue_t)queue;
414{
415    if (queue) {
416        sr_dispatch_retain(queue);
417    }
418
419    if (_delegateDispatchQueue) {
420        sr_dispatch_release(_delegateDispatchQueue);
421    }
422
423    _delegateDispatchQueue = queue;
424}
425
426- (BOOL)_checkHandshake:(CFHTTPMessageRef)httpMessage;
427{
428    NSString *acceptHeader = CFBridgingRelease(CFHTTPMessageCopyHeaderFieldValue(httpMessage, CFSTR("Sec-WebSocket-Accept")));
429
430    if (acceptHeader == nil) {
431        return NO;
432    }
433
434    NSString *concattedString = [_secKey stringByAppendingString:SRWebSocketAppendToSecKeyString];
435    NSString *expectedAccept = [concattedString stringBySHA1ThenBase64Encoding];
436
437    return [acceptHeader isEqualToString:expectedAccept];
438}
439
440- (void)_HTTPHeadersDidFinish;
441{
442    NSInteger responseCode = CFHTTPMessageGetResponseStatusCode(_receivedHTTPHeaders);
443
444    if (responseCode >= 400) {
445        SRFastLog(@"Request failed with response code %d", responseCode);
446        [self _failWithError:[NSError errorWithDomain:SRWebSocketErrorDomain code:2132 userInfo:@{NSLocalizedDescriptionKey:[NSString stringWithFormat:@"received bad response code from server %ld", (long)responseCode], SRHTTPResponseErrorKey:@(responseCode)}]];
447        return;
448    }
449
450    if(![self _checkHandshake:_receivedHTTPHeaders]) {
451        [self _failWithError:[NSError errorWithDomain:SRWebSocketErrorDomain code:2133 userInfo:[NSDictionary dictionaryWithObject:[NSString stringWithFormat:@"Invalid Sec-WebSocket-Accept response"] forKey:NSLocalizedDescriptionKey]]];
452        return;
453    }
454
455    NSString *negotiatedProtocol = CFBridgingRelease(CFHTTPMessageCopyHeaderFieldValue(_receivedHTTPHeaders, CFSTR("Sec-WebSocket-Protocol")));
456    if (negotiatedProtocol) {
457        // Make sure we requested the protocol
458        if ([_requestedProtocols indexOfObject:negotiatedProtocol] == NSNotFound) {
459            [self _failWithError:[NSError errorWithDomain:SRWebSocketErrorDomain code:2133 userInfo:[NSDictionary dictionaryWithObject:[NSString stringWithFormat:@"Server specified Sec-WebSocket-Protocol that wasn't requested"] forKey:NSLocalizedDescriptionKey]]];
460            return;
461        }
462
463        _protocol = negotiatedProtocol;
464    }
465
466    self.readyState = SR_OPEN;
467
468    if (!_didFail) {
469        [self _readFrameNew];
470    }
471
472    [self _performDelegateBlock:^{
473        if ([self.delegate respondsToSelector:@selector(webSocketDidOpen:)]) {
474            [self.delegate webSocketDidOpen:self];
475        };
476    }];
477}
478
479
480- (void)_readHTTPHeader;
481{
482    if (_receivedHTTPHeaders == NULL) {
483        _receivedHTTPHeaders = CFHTTPMessageCreateEmpty(NULL, NO);
484    }
485
486    [self _readUntilHeaderCompleteWithCallback:^(SRWebSocket *self,  NSData *data) {
487        CFHTTPMessageAppendBytes(_receivedHTTPHeaders, (const UInt8 *)data.bytes, data.length);
488
489        if (CFHTTPMessageIsHeaderComplete(_receivedHTTPHeaders)) {
490            SRFastLog(@"Finished reading headers %@", CFBridgingRelease(CFHTTPMessageCopyAllHeaderFields(_receivedHTTPHeaders)));
491            [self _HTTPHeadersDidFinish];
492        } else {
493            [self _readHTTPHeader];
494        }
495    }];
496}
497
498- (void)didConnect
499{
500    SRFastLog(@"Connected");
501    CFHTTPMessageRef request = CFHTTPMessageCreateRequest(NULL, CFSTR("GET"), (__bridge CFURLRef)_url, kCFHTTPVersion1_1);
502
503    // Set host first so it defaults
504    CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Host"), (__bridge CFStringRef)(_url.port ? [NSString stringWithFormat:@"%@:%@", _url.host, _url.port] : _url.host));
505
506    NSMutableData *keyBytes = [[NSMutableData alloc] initWithLength:16];
507    BOOL success = !SecRandomCopyBytes(kSecRandomDefault, keyBytes.length, keyBytes.mutableBytes);
508    assert(success);
509
510    if ([keyBytes respondsToSelector:@selector(base64EncodedStringWithOptions:)]) {
511        _secKey = [keyBytes base64EncodedStringWithOptions:0];
512    } else {
513        _secKey = [keyBytes base64Encoding];
514    }
515
516    assert([_secKey length] == 24);
517
518    CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Upgrade"), CFSTR("websocket"));
519    CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Connection"), CFSTR("Upgrade"));
520    CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Sec-WebSocket-Key"), (__bridge CFStringRef)_secKey);
521    CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Sec-WebSocket-Version"), (__bridge CFStringRef)[NSString stringWithFormat:@"%ld", (long)_webSocketVersion]);
522
523    CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Origin"), (__bridge CFStringRef)_url.SR_origin);
524
525    if (_requestedProtocols) {
526        CFHTTPMessageSetHeaderFieldValue(request, CFSTR("Sec-WebSocket-Protocol"), (__bridge CFStringRef)[_requestedProtocols componentsJoinedByString:@", "]);
527    }
528
529    [_urlRequest.allHTTPHeaderFields enumerateKeysAndObjectsUsingBlock:^(id key, id obj, BOOL *stop) {
530        CFHTTPMessageSetHeaderFieldValue(request, (__bridge CFStringRef)key, (__bridge CFStringRef)obj);
531    }];
532
533    NSData *message = CFBridgingRelease(CFHTTPMessageCopySerializedMessage(request));
534
535    CFRelease(request);
536
537    [self _writeData:message];
538    [self _readHTTPHeader];
539}
540
541- (void)_initializeStreams;
542{
543    assert(_url.port.unsignedIntValue <= UINT32_MAX);
544    uint32_t port = _url.port.unsignedIntValue;
545    if (port == 0) {
546        if (!_secure) {
547            port = 80;
548        } else {
549            port = 443;
550        }
551    }
552    NSString *host = _url.host;
553
554    CFReadStreamRef readStream = NULL;
555    CFWriteStreamRef writeStream = NULL;
556
557    CFStreamCreatePairWithSocketToHost(NULL, (__bridge CFStringRef)host, port, &readStream, &writeStream);
558
559    _outputStream = CFBridgingRelease(writeStream);
560    _inputStream = CFBridgingRelease(readStream);
561
562
563    if (_secure) {
564        NSMutableDictionary *SSLOptions = [[NSMutableDictionary alloc] init];
565
566        [_outputStream setProperty:(__bridge id)kCFStreamSocketSecurityLevelNegotiatedSSL forKey:(__bridge id)kCFStreamPropertySocketSecurityLevel];
567
568        // If we're using pinned certs, don't validate the certificate chain
569        if ([_urlRequest SR_SSLPinnedCertificates].count) {
570            [SSLOptions setValue:[NSNumber numberWithBool:NO] forKey:(__bridge id)kCFStreamSSLValidatesCertificateChain];
571        }
572
573#ifdef DEBUG
574        [SSLOptions setValue:[NSNumber numberWithBool:NO] forKey:(__bridge id)kCFStreamSSLValidatesCertificateChain];
575        NSLog(@"SocketRocket: In debug mode.  Allowing connection to any root cert");
576#endif
577
578        [_outputStream setProperty:SSLOptions
579                            forKey:(__bridge id)kCFStreamPropertySSLSettings];
580    }
581
582    _inputStream.delegate = self;
583    _outputStream.delegate = self;
584}
585
586- (void)_connect;
587{
588    if (!_scheduledRunloops.count) {
589        [self scheduleInRunLoop:[NSRunLoop SR_networkRunLoop] forMode:NSDefaultRunLoopMode];
590    }
591
592
593    [_outputStream open];
594    [_inputStream open];
595}
596
597- (void)scheduleInRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode;
598{
599    [_outputStream scheduleInRunLoop:aRunLoop forMode:mode];
600    [_inputStream scheduleInRunLoop:aRunLoop forMode:mode];
601
602    [_scheduledRunloops addObject:@[aRunLoop, mode]];
603}
604
605- (void)unscheduleFromRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode;
606{
607    [_outputStream removeFromRunLoop:aRunLoop forMode:mode];
608    [_inputStream removeFromRunLoop:aRunLoop forMode:mode];
609
610    [_scheduledRunloops removeObject:@[aRunLoop, mode]];
611}
612
613- (void)close;
614{
615    [self closeWithCode:SRStatusCodeNormal reason:nil];
616}
617
618- (void)closeWithCode:(NSInteger)code reason:(NSString *)reason;
619{
620    assert(code);
621    dispatch_async(_workQueue, ^{
622        if (self.readyState == SR_CLOSING || self.readyState == SR_CLOSED) {
623            return;
624        }
625
626        BOOL wasConnecting = self.readyState == SR_CONNECTING;
627
628        self.readyState = SR_CLOSING;
629
630        SRFastLog(@"Closing with code %d reason %@", code, reason);
631
632        if (wasConnecting) {
633            [self _disconnect];
634            return;
635        }
636
637        size_t maxMsgSize = [reason maximumLengthOfBytesUsingEncoding:NSUTF8StringEncoding];
638        NSMutableData *mutablePayload = [[NSMutableData alloc] initWithLength:sizeof(uint16_t) + maxMsgSize];
639        NSData *payload = mutablePayload;
640
641        ((uint16_t *)mutablePayload.mutableBytes)[0] = EndianU16_BtoN(code);
642
643        if (reason) {
644            NSRange remainingRange = {0};
645
646            NSUInteger usedLength = 0;
647
648            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];
649
650            assert(success);
651            assert(remainingRange.length == 0);
652
653            if (usedLength != maxMsgSize) {
654                payload = [payload subdataWithRange:NSMakeRange(0, usedLength + sizeof(uint16_t))];
655            }
656        }
657
658
659        [self _sendFrameWithOpcode:SROpCodeConnectionClose data:payload];
660    });
661}
662
663- (void)_closeWithProtocolError:(NSString *)message;
664{
665    // Need to shunt this on the _callbackQueue first to see if they received any messages
666    [self _performDelegateBlock:^{
667        [self closeWithCode:SRStatusCodeProtocolError reason:message];
668        dispatch_async(_workQueue, ^{
669            [self _disconnect];
670        });
671    }];
672}
673
674- (void)_failWithError:(NSError *)error;
675{
676    dispatch_async(_workQueue, ^{
677        if (self.readyState != SR_CLOSED) {
678            _failed = YES;
679            [self _performDelegateBlock:^{
680                if ([self.delegate respondsToSelector:@selector(webSocket:didFailWithError:)]) {
681                    [self.delegate webSocket:self didFailWithError:error];
682                }
683            }];
684
685            self.readyState = SR_CLOSED;
686            _selfRetain = nil;
687
688            SRFastLog(@"Failing with error %@", error.localizedDescription);
689
690            [self _disconnect];
691        }
692    });
693}
694
695- (void)_writeData:(NSData *)data;
696{
697    [self assertOnWorkQueue];
698
699    if (_closeWhenFinishedWriting) {
700            return;
701    }
702    [_outputBuffer appendData:data];
703    [self _pumpWriting];
704}
705
706- (void)send:(id)data;
707{
708    NSAssert(self.readyState != SR_CONNECTING, @"Invalid State: Cannot call send: until connection is open");
709    // TODO: maybe not copy this for performance
710    data = [data copy];
711    dispatch_async(_workQueue, ^{
712        if ([data isKindOfClass:[NSString class]]) {
713            [self _sendFrameWithOpcode:SROpCodeTextFrame data:[(NSString *)data dataUsingEncoding:NSUTF8StringEncoding]];
714        } else if ([data isKindOfClass:[NSData class]]) {
715            [self _sendFrameWithOpcode:SROpCodeBinaryFrame data:data];
716        } else if (data == nil) {
717            [self _sendFrameWithOpcode:SROpCodeTextFrame data:data];
718        } else {
719            assert(NO);
720        }
721    });
722}
723
724- (void)sendPing:(NSData *)data;
725{
726    NSAssert(self.readyState == SR_OPEN, @"Invalid State: Cannot call send: until connection is open");
727    // TODO: maybe not copy this for performance
728    data = [data copy] ?: [NSData data]; // It's okay for a ping to be empty
729    dispatch_async(_workQueue, ^{
730        [self _sendFrameWithOpcode:SROpCodePing data:data];
731    });
732}
733
734- (void)handlePing:(NSData *)pingData;
735{
736    // Need to pingpong this off _callbackQueue first to make sure messages happen in order
737    [self _performDelegateBlock:^{
738        dispatch_async(_workQueue, ^{
739            [self _sendFrameWithOpcode:SROpCodePong data:pingData];
740        });
741    }];
742}
743
744- (void)handlePong:(NSData *)pongData;
745{
746    SRFastLog(@"Received pong");
747    [self _performDelegateBlock:^{
748        if ([self.delegate respondsToSelector:@selector(webSocket:didReceivePong:)]) {
749            [self.delegate webSocket:self didReceivePong:pongData];
750        }
751    }];
752}
753
754- (void)_handleMessage:(id)message
755{
756    SRFastLog(@"Received message");
757    [self _performDelegateBlock:^{
758        [self.delegate webSocket:self didReceiveMessage:message];
759    }];
760}
761
762
763static inline BOOL closeCodeIsValid(int closeCode) {
764    if (closeCode < 1000) {
765        return NO;
766    }
767
768    if (closeCode >= 1000 && closeCode <= 1011) {
769        if (closeCode == 1004 ||
770            closeCode == 1005 ||
771            closeCode == 1006) {
772            return NO;
773        }
774        return YES;
775    }
776
777    if (closeCode >= 3000 && closeCode <= 3999) {
778        return YES;
779    }
780
781    if (closeCode >= 4000 && closeCode <= 4999) {
782        return YES;
783    }
784
785    return NO;
786}
787
788//  Note from RFC:
789//
790//  If there is a body, the first two
791//  bytes of the body MUST be a 2-byte unsigned integer (in network byte
792//  order) representing a status code with value /code/ defined in
793//  Section 7.4.  Following the 2-byte integer the body MAY contain UTF-8
794//  encoded data with value /reason/, the interpretation of which is not
795//  defined by this specification.
796
797- (void)handleCloseWithData:(NSData *)data;
798{
799    size_t dataSize = data.length;
800    __block uint16_t closeCode = 0;
801
802    SRFastLog(@"Received close frame");
803
804    if (dataSize == 1) {
805        // TODO handle error
806        [self _closeWithProtocolError:@"Payload for close must be larger than 2 bytes"];
807        return;
808    } else if (dataSize >= 2) {
809        [data getBytes:&closeCode length:sizeof(closeCode)];
810        _closeCode = EndianU16_BtoN(closeCode);
811        if (!closeCodeIsValid(_closeCode)) {
812            [self _closeWithProtocolError:[NSString stringWithFormat:@"Cannot have close code of %d", _closeCode]];
813            return;
814        }
815        if (dataSize > 2) {
816            _closeReason = [[NSString alloc] initWithData:[data subdataWithRange:NSMakeRange(2, dataSize - 2)] encoding:NSUTF8StringEncoding];
817            if (!_closeReason) {
818                [self _closeWithProtocolError:@"Close reason MUST be valid UTF-8"];
819                return;
820            }
821        }
822    } else {
823        _closeCode = SRStatusNoStatusReceived;
824    }
825
826    [self assertOnWorkQueue];
827
828    if (self.readyState == SR_OPEN) {
829        [self closeWithCode:1000 reason:nil];
830    }
831    dispatch_async(_workQueue, ^{
832        [self _disconnect];
833    });
834}
835
836- (void)_disconnect;
837{
838    [self assertOnWorkQueue];
839    SRFastLog(@"Trying to disconnect");
840    _closeWhenFinishedWriting = YES;
841    [self _pumpWriting];
842}
843
844- (void)_handleFrameWithData:(NSData *)frameData opCode:(NSInteger)opcode;
845{
846    // Check that the current data is valid UTF8
847
848    BOOL isControlFrame = (opcode == SROpCodePing || opcode == SROpCodePong || opcode == SROpCodeConnectionClose);
849    if (!isControlFrame) {
850        [self _readFrameNew];
851    } else {
852        dispatch_async(_workQueue, ^{
853            [self _readFrameContinue];
854        });
855    }
856
857    switch (opcode) {
858        case SROpCodeTextFrame: {
859            NSString *str = [[NSString alloc] initWithData:frameData encoding:NSUTF8StringEncoding];
860            if (str == nil && frameData) {
861                [self closeWithCode:SRStatusCodeInvalidUTF8 reason:@"Text frames must be valid UTF-8"];
862                dispatch_async(_workQueue, ^{
863                    [self _disconnect];
864                });
865
866                return;
867            }
868            [self _handleMessage:str];
869            break;
870        }
871        case SROpCodeBinaryFrame:
872            [self _handleMessage:[frameData copy]];
873            break;
874        case SROpCodeConnectionClose:
875            [self handleCloseWithData:frameData];
876            break;
877        case SROpCodePing:
878            [self handlePing:frameData];
879            break;
880        case SROpCodePong:
881            [self handlePong:frameData];
882            break;
883        default:
884            [self _closeWithProtocolError:[NSString stringWithFormat:@"Unknown opcode %ld", (long)opcode]];
885            // TODO: Handle invalid opcode
886            break;
887    }
888}
889
890- (void)_handleFrameHeader:(frame_header)frame_header curData:(NSData *)curData;
891{
892    assert(frame_header.opcode != 0);
893
894    if (self.readyState != SR_OPEN) {
895        return;
896    }
897
898
899    BOOL isControlFrame = (frame_header.opcode == SROpCodePing || frame_header.opcode == SROpCodePong || frame_header.opcode == SROpCodeConnectionClose);
900
901    if (isControlFrame && !frame_header.fin) {
902        [self _closeWithProtocolError:@"Fragmented control frames not allowed"];
903        return;
904    }
905
906    if (isControlFrame && frame_header.payload_length >= 126) {
907        [self _closeWithProtocolError:@"Control frames cannot have payloads larger than 126 bytes"];
908        return;
909    }
910
911    if (!isControlFrame) {
912        _currentFrameOpcode = frame_header.opcode;
913        _currentFrameCount += 1;
914    }
915
916    if (frame_header.payload_length == 0) {
917        if (isControlFrame) {
918            [self _handleFrameWithData:curData opCode:frame_header.opcode];
919        } else {
920            if (frame_header.fin) {
921                [self _handleFrameWithData:_currentFrameData opCode:frame_header.opcode];
922            } else {
923                // TODO add assert that opcode is not a control;
924                [self _readFrameContinue];
925            }
926        }
927    } else {
928        assert(frame_header.payload_length <= SIZE_T_MAX);
929        [self _addConsumerWithDataLength:(size_t)frame_header.payload_length callback:^(SRWebSocket *self, NSData *newData) {
930            if (isControlFrame) {
931                [self _handleFrameWithData:newData opCode:frame_header.opcode];
932            } else {
933                if (frame_header.fin) {
934                    [self _handleFrameWithData:self->_currentFrameData opCode:frame_header.opcode];
935                } else {
936                    // TODO add assert that opcode is not a control;
937                    [self _readFrameContinue];
938                }
939
940            }
941        } readToCurrentFrame:!isControlFrame unmaskBytes:frame_header.masked];
942    }
943}
944
945/* From RFC:
946
947 0                   1                   2                   3
948 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
949 +-+-+-+-+-------+-+-------------+-------------------------------+
950 |F|R|R|R| opcode|M| Payload len |    Extended payload length    |
951 |I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
952 |N|V|V|V|       |S|             |   (if payload len==126/127)   |
953 | |1|2|3|       |K|             |                               |
954 +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
955 |     Extended payload length continued, if payload len == 127  |
956 + - - - - - - - - - - - - - - - +-------------------------------+
957 |                               |Masking-key, if MASK set to 1  |
958 +-------------------------------+-------------------------------+
959 | Masking-key (continued)       |          Payload Data         |
960 +-------------------------------- - - - - - - - - - - - - - - - +
961 :                     Payload Data continued ...                :
962 + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
963 |                     Payload Data continued ...                |
964 +---------------------------------------------------------------+
965 */
966
967static const uint8_t SRFinMask          = 0x80;
968static const uint8_t SROpCodeMask       = 0x0F;
969static const uint8_t SRRsvMask          = 0x70;
970static const uint8_t SRMaskMask         = 0x80;
971static const uint8_t SRPayloadLenMask   = 0x7F;
972
973
974- (void)_readFrameContinue;
975{
976    assert((_currentFrameCount == 0 && _currentFrameOpcode == 0) || (_currentFrameCount > 0 && _currentFrameOpcode > 0));
977
978    [self _addConsumerWithDataLength:2 callback:^(SRWebSocket *self, NSData *data) {
979        __block frame_header header = {0};
980
981        const uint8_t *headerBuffer = data.bytes;
982        assert(data.length >= 2);
983
984        if (headerBuffer[0] & SRRsvMask) {
985            [self _closeWithProtocolError:@"Server used RSV bits"];
986            return;
987        }
988
989        uint8_t receivedOpcode = (SROpCodeMask & headerBuffer[0]);
990
991        BOOL isControlFrame = (receivedOpcode == SROpCodePing || receivedOpcode == SROpCodePong || receivedOpcode == SROpCodeConnectionClose);
992
993        if (!isControlFrame && receivedOpcode != 0 && self->_currentFrameCount > 0) {
994            [self _closeWithProtocolError:@"all data frames after the initial data frame must have opcode 0"];
995            return;
996        }
997
998        if (receivedOpcode == 0 && self->_currentFrameCount == 0) {
999            [self _closeWithProtocolError:@"cannot continue a message"];
1000            return;
1001        }
1002
1003        header.opcode = receivedOpcode == 0 ? self->_currentFrameOpcode : receivedOpcode;
1004
1005        header.fin = !!(SRFinMask & headerBuffer[0]);
1006
1007
1008        header.masked = !!(SRMaskMask & headerBuffer[1]);
1009        header.payload_length = SRPayloadLenMask & headerBuffer[1];
1010
1011        headerBuffer = NULL;
1012
1013        if (header.masked) {
1014            [self _closeWithProtocolError:@"Client must receive unmasked data"];
1015        }
1016
1017        size_t extra_bytes_needed = header.masked ? sizeof(_currentReadMaskKey) : 0;
1018
1019        if (header.payload_length == 126) {
1020            extra_bytes_needed += sizeof(uint16_t);
1021        } else if (header.payload_length == 127) {
1022            extra_bytes_needed += sizeof(uint64_t);
1023        }
1024
1025        if (extra_bytes_needed == 0) {
1026            [self _handleFrameHeader:header curData:self->_currentFrameData];
1027        } else {
1028            [self _addConsumerWithDataLength:extra_bytes_needed callback:^(SRWebSocket *self, NSData *data) {
1029                size_t mapped_size = data.length;
1030                const void *mapped_buffer = data.bytes;
1031                size_t offset = 0;
1032
1033                if (header.payload_length == 126) {
1034                    assert(mapped_size >= sizeof(uint16_t));
1035                    uint16_t newLen = EndianU16_BtoN(*(uint16_t *)(mapped_buffer));
1036                    header.payload_length = newLen;
1037                    offset += sizeof(uint16_t);
1038                } else if (header.payload_length == 127) {
1039                    assert(mapped_size >= sizeof(uint64_t));
1040                    header.payload_length = EndianU64_BtoN(*(uint64_t *)(mapped_buffer));
1041                    offset += sizeof(uint64_t);
1042                } else {
1043                    assert(header.payload_length < 126 && header.payload_length >= 0);
1044                }
1045
1046
1047                if (header.masked) {
1048                    assert(mapped_size >= sizeof(_currentReadMaskOffset) + offset);
1049                    memcpy(self->_currentReadMaskKey, ((uint8_t *)mapped_buffer) + offset, sizeof(self->_currentReadMaskKey));
1050                }
1051
1052                [self _handleFrameHeader:header curData:self->_currentFrameData];
1053            } readToCurrentFrame:NO unmaskBytes:NO];
1054        }
1055    } readToCurrentFrame:NO unmaskBytes:NO];
1056}
1057
1058- (void)_readFrameNew;
1059{
1060    dispatch_async(_workQueue, ^{
1061        [_currentFrameData setLength:0];
1062
1063        _currentFrameOpcode = 0;
1064        _currentFrameCount = 0;
1065        _readOpCount = 0;
1066        _currentStringScanPosition = 0;
1067
1068        [self _readFrameContinue];
1069    });
1070}
1071
1072- (void)_pumpWriting;
1073{
1074    [self assertOnWorkQueue];
1075
1076    NSUInteger dataLength = _outputBuffer.length;
1077    if (dataLength - _outputBufferOffset > 0 && _outputStream.hasSpaceAvailable) {
1078        NSInteger bytesWritten = [_outputStream write:_outputBuffer.bytes + _outputBufferOffset maxLength:dataLength - _outputBufferOffset];
1079        if (bytesWritten == -1) {
1080            [self _failWithError:[NSError errorWithDomain:SRWebSocketErrorDomain code:2145 userInfo:[NSDictionary dictionaryWithObject:@"Error writing to stream" forKey:NSLocalizedDescriptionKey]]];
1081             return;
1082        }
1083
1084        _outputBufferOffset += bytesWritten;
1085
1086        if (_outputBufferOffset > 4096 && _outputBufferOffset > (_outputBuffer.length >> 1)) {
1087            _outputBuffer = [[NSMutableData alloc] initWithBytes:(char *)_outputBuffer.bytes + _outputBufferOffset length:_outputBuffer.length - _outputBufferOffset];
1088            _outputBufferOffset = 0;
1089        }
1090    }
1091
1092    if (_closeWhenFinishedWriting &&
1093        _outputBuffer.length - _outputBufferOffset == 0 &&
1094        (_inputStream.streamStatus != NSStreamStatusNotOpen &&
1095         _inputStream.streamStatus != NSStreamStatusClosed) &&
1096        !_sentClose) {
1097        _sentClose = YES;
1098
1099        [_outputStream close];
1100        [_inputStream close];
1101
1102
1103        for (NSArray *runLoop in [_scheduledRunloops copy]) {
1104            [self unscheduleFromRunLoop:[runLoop objectAtIndex:0] forMode:[runLoop objectAtIndex:1]];
1105        }
1106
1107        if (!_failed) {
1108            [self _performDelegateBlock:^{
1109                if ([self.delegate respondsToSelector:@selector(webSocket:didCloseWithCode:reason:wasClean:)]) {
1110                    [self.delegate webSocket:self didCloseWithCode:_closeCode reason:_closeReason wasClean:YES];
1111                }
1112            }];
1113        }
1114
1115        _selfRetain = nil;
1116    }
1117}
1118
1119- (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback;
1120{
1121    [self assertOnWorkQueue];
1122    [self _addConsumerWithScanner:consumer callback:callback dataLength:0];
1123}
1124
1125- (void)_addConsumerWithDataLength:(size_t)dataLength callback:(data_callback)callback readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes;
1126{
1127    [self assertOnWorkQueue];
1128    assert(dataLength);
1129
1130    [_consumers addObject:[_consumerPool consumerWithScanner:nil handler:callback bytesNeeded:dataLength readToCurrentFrame:readToCurrentFrame unmaskBytes:unmaskBytes]];
1131    [self _pumpScanner];
1132}
1133
1134- (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback dataLength:(size_t)dataLength;
1135{
1136    [self assertOnWorkQueue];
1137    [_consumers addObject:[_consumerPool consumerWithScanner:consumer handler:callback bytesNeeded:dataLength readToCurrentFrame:NO unmaskBytes:NO]];
1138    [self _pumpScanner];
1139}
1140
1141
1142static const char CRLFCRLFBytes[] = {'\r', '\n', '\r', '\n'};
1143
1144- (void)_readUntilHeaderCompleteWithCallback:(data_callback)dataHandler;
1145{
1146    [self _readUntilBytes:CRLFCRLFBytes length:sizeof(CRLFCRLFBytes) callback:dataHandler];
1147}
1148
1149- (void)_readUntilBytes:(const void *)bytes length:(size_t)length callback:(data_callback)dataHandler;
1150{
1151    // TODO optimize so this can continue from where we last searched
1152    stream_scanner consumer = ^size_t(NSData *data) {
1153        __block size_t found_size = 0;
1154        __block size_t match_count = 0;
1155
1156        size_t size = data.length;
1157        const unsigned char *buffer = data.bytes;
1158        for (size_t i = 0; i < size; i++ ) {
1159            if (((const unsigned char *)buffer)[i] == ((const unsigned char *)bytes)[match_count]) {
1160                match_count += 1;
1161                if (match_count == length) {
1162                    found_size = i + 1;
1163                    break;
1164                }
1165            } else {
1166                match_count = 0;
1167            }
1168        }
1169        return found_size;
1170    };
1171    [self _addConsumerWithScanner:consumer callback:dataHandler];
1172}
1173
1174
1175// Returns true if did work
1176- (BOOL)_innerPumpScanner {
1177
1178    BOOL didWork = NO;
1179
1180    if (self.readyState >= SR_CLOSING) {
1181        return didWork;
1182    }
1183
1184    if (!_consumers.count) {
1185        return didWork;
1186    }
1187
1188    size_t curSize = _readBuffer.length - _readBufferOffset;
1189    if (!curSize) {
1190        return didWork;
1191    }
1192
1193    SRIOConsumer *consumer = [_consumers objectAtIndex:0];
1194
1195    size_t bytesNeeded = consumer.bytesNeeded;
1196
1197    size_t foundSize = 0;
1198    if (consumer.consumer) {
1199        NSData *tempView = [NSData dataWithBytesNoCopy:(char *)_readBuffer.bytes + _readBufferOffset length:_readBuffer.length - _readBufferOffset freeWhenDone:NO];
1200        foundSize = consumer.consumer(tempView);
1201    } else {
1202        assert(consumer.bytesNeeded);
1203        if (curSize >= bytesNeeded) {
1204            foundSize = bytesNeeded;
1205        } else if (consumer.readToCurrentFrame) {
1206            foundSize = curSize;
1207        }
1208    }
1209
1210    NSData *slice = nil;
1211    if (consumer.readToCurrentFrame || foundSize) {
1212        NSRange sliceRange = NSMakeRange(_readBufferOffset, foundSize);
1213        slice = [_readBuffer subdataWithRange:sliceRange];
1214
1215        _readBufferOffset += foundSize;
1216
1217        if (_readBufferOffset > 4096 && _readBufferOffset > (_readBuffer.length >> 1)) {
1218            _readBuffer = [[NSMutableData alloc] initWithBytes:(char *)_readBuffer.bytes + _readBufferOffset length:_readBuffer.length - _readBufferOffset];            _readBufferOffset = 0;
1219        }
1220
1221        if (consumer.unmaskBytes) {
1222            NSMutableData *mutableSlice = [slice mutableCopy];
1223
1224            NSUInteger len = mutableSlice.length;
1225            uint8_t *bytes = mutableSlice.mutableBytes;
1226
1227            for (NSUInteger i = 0; i < len; i++) {
1228                bytes[i] = bytes[i] ^ _currentReadMaskKey[_currentReadMaskOffset % sizeof(_currentReadMaskKey)];
1229                _currentReadMaskOffset += 1;
1230            }
1231
1232            slice = mutableSlice;
1233        }
1234
1235        if (consumer.readToCurrentFrame) {
1236            [_currentFrameData appendData:slice];
1237
1238            _readOpCount += 1;
1239
1240            if (_currentFrameOpcode == SROpCodeTextFrame) {
1241                // Validate UTF8 stuff.
1242                size_t currentDataSize = _currentFrameData.length;
1243                if (_currentFrameOpcode == SROpCodeTextFrame && currentDataSize > 0) {
1244                    // TODO: Optimize the crap out of this.  Don't really have to copy all the data each time
1245
1246                    size_t scanSize = currentDataSize - _currentStringScanPosition;
1247
1248                    NSData *scan_data = [_currentFrameData subdataWithRange:NSMakeRange(_currentStringScanPosition, scanSize)];
1249                    int32_t valid_utf8_size = validate_dispatch_data_partial_string(scan_data);
1250
1251                    if (valid_utf8_size == -1) {
1252                        [self closeWithCode:SRStatusCodeInvalidUTF8 reason:@"Text frames must be valid UTF-8"];
1253                        dispatch_async(_workQueue, ^{
1254                            [self _disconnect];
1255                        });
1256                        return didWork;
1257                    } else {
1258                        _currentStringScanPosition += valid_utf8_size;
1259                    }
1260                }
1261
1262            }
1263
1264            consumer.bytesNeeded -= foundSize;
1265
1266            if (consumer.bytesNeeded == 0) {
1267                [_consumers removeObjectAtIndex:0];
1268                consumer.handler(self, nil);
1269                [_consumerPool returnConsumer:consumer];
1270                didWork = YES;
1271            }
1272        } else if (foundSize) {
1273            [_consumers removeObjectAtIndex:0];
1274            consumer.handler(self, slice);
1275            [_consumerPool returnConsumer:consumer];
1276            didWork = YES;
1277        }
1278    }
1279    return didWork;
1280}
1281
1282-(void)_pumpScanner;
1283{
1284    [self assertOnWorkQueue];
1285
1286    if (!_isPumping) {
1287        _isPumping = YES;
1288    } else {
1289        return;
1290    }
1291
1292    while ([self _innerPumpScanner]) {
1293
1294    }
1295
1296    _isPumping = NO;
1297}
1298
1299//#define NOMASK
1300
1301static const size_t SRFrameHeaderOverhead = 32;
1302
1303- (void)_sendFrameWithOpcode:(SROpCode)opcode data:(id)data;
1304{
1305    [self assertOnWorkQueue];
1306
1307    if (nil == data) {
1308        return;
1309    }
1310
1311    NSAssert([data isKindOfClass:[NSData class]] || [data isKindOfClass:[NSString class]], @"NSString or NSData");
1312
1313    size_t payloadLength = [data isKindOfClass:[NSString class]] ? [(NSString *)data lengthOfBytesUsingEncoding:NSUTF8StringEncoding] : [data length];
1314
1315    NSMutableData *frame = [[NSMutableData alloc] initWithLength:payloadLength + SRFrameHeaderOverhead];
1316    if (!frame) {
1317        [self closeWithCode:SRStatusCodeMessageTooBig reason:@"Message too big"];
1318        return;
1319    }
1320    uint8_t *frame_buffer = (uint8_t *)[frame mutableBytes];
1321
1322    // set fin
1323    frame_buffer[0] = SRFinMask | opcode;
1324
1325    BOOL useMask = YES;
1326#ifdef NOMASK
1327    useMask = NO;
1328#endif
1329
1330    if (useMask) {
1331    // set the mask and header
1332        frame_buffer[1] |= SRMaskMask;
1333    }
1334
1335    size_t frame_buffer_size = 2;
1336
1337    const uint8_t *unmasked_payload = NULL;
1338    if ([data isKindOfClass:[NSData class]]) {
1339        unmasked_payload = (uint8_t *)[data bytes];
1340    } else if ([data isKindOfClass:[NSString class]]) {
1341        unmasked_payload =  (const uint8_t *)[data UTF8String];
1342    } else {
1343        return;
1344    }
1345
1346    if (payloadLength < 126) {
1347        frame_buffer[1] |= payloadLength;
1348    } else if (payloadLength <= UINT16_MAX) {
1349        frame_buffer[1] |= 126;
1350        *((uint16_t *)(frame_buffer + frame_buffer_size)) = EndianU16_BtoN((uint16_t)payloadLength);
1351        frame_buffer_size += sizeof(uint16_t);
1352    } else {
1353        frame_buffer[1] |= 127;
1354        *((uint64_t *)(frame_buffer + frame_buffer_size)) = EndianU64_BtoN((uint64_t)payloadLength);
1355        frame_buffer_size += sizeof(uint64_t);
1356    }
1357
1358    if (!useMask) {
1359        for (size_t i = 0; i < payloadLength; i++) {
1360            frame_buffer[frame_buffer_size] = unmasked_payload[i];
1361            frame_buffer_size += 1;
1362        }
1363    } else {
1364        uint8_t *mask_key = frame_buffer + frame_buffer_size;
1365        BOOL success = !SecRandomCopyBytes(kSecRandomDefault, sizeof(uint32_t), (uint8_t *)mask_key);
1366        assert(success);
1367        frame_buffer_size += sizeof(uint32_t);
1368
1369        // TODO: could probably optimize this with SIMD
1370        for (size_t i = 0; i < payloadLength; i++) {
1371            frame_buffer[frame_buffer_size] = unmasked_payload[i] ^ mask_key[i % sizeof(uint32_t)];
1372            frame_buffer_size += 1;
1373        }
1374    }
1375
1376    assert(frame_buffer_size <= [frame length]);
1377    frame.length = frame_buffer_size;
1378
1379    [self _writeData:frame];
1380}
1381
1382- (void)stream:(NSStream *)aStream handleEvent:(NSStreamEvent)eventCode;
1383{
1384    if (_secure && !_pinnedCertFound && (eventCode == NSStreamEventHasBytesAvailable || eventCode == NSStreamEventHasSpaceAvailable)) {
1385
1386        NSArray *sslCerts = [_urlRequest SR_SSLPinnedCertificates];
1387        if (sslCerts) {
1388            SecTrustRef secTrust = (__bridge SecTrustRef)[aStream propertyForKey:(__bridge id)kCFStreamPropertySSLPeerTrust];
1389            if (secTrust) {
1390                NSInteger numCerts = SecTrustGetCertificateCount(secTrust);
1391                for (NSInteger i = 0; i < numCerts && !_pinnedCertFound; i++) {
1392                    SecCertificateRef cert = SecTrustGetCertificateAtIndex(secTrust, i);
1393                    NSData *certData = CFBridgingRelease(SecCertificateCopyData(cert));
1394
1395                    for (id ref in sslCerts) {
1396                        SecCertificateRef trustedCert = (__bridge SecCertificateRef)ref;
1397                        NSData *trustedCertData = CFBridgingRelease(SecCertificateCopyData(trustedCert));
1398
1399                        if ([trustedCertData isEqualToData:certData]) {
1400                            _pinnedCertFound = YES;
1401                            break;
1402                        }
1403                    }
1404                }
1405            }
1406
1407            if (!_pinnedCertFound) {
1408                dispatch_async(_workQueue, ^{
1409                    [self _failWithError:[NSError errorWithDomain:SRWebSocketErrorDomain code:23556 userInfo:[NSDictionary dictionaryWithObject:[NSString stringWithFormat:@"Invalid server cert"] forKey:NSLocalizedDescriptionKey]]];
1410                });
1411                return;
1412            }
1413        }
1414    }
1415
1416    dispatch_async(_workQueue, ^{
1417        switch (eventCode) {
1418            case NSStreamEventOpenCompleted: {
1419                SRFastLog(@"NSStreamEventOpenCompleted %@", aStream);
1420                if (self.readyState >= SR_CLOSING) {
1421                    return;
1422                }
1423                assert(_readBuffer);
1424
1425                if (self.readyState == SR_CONNECTING && aStream == _inputStream) {
1426                    [self didConnect];
1427                }
1428                [self _pumpWriting];
1429                [self _pumpScanner];
1430                break;
1431            }
1432
1433            case NSStreamEventErrorOccurred: {
1434                SRFastLog(@"NSStreamEventErrorOccurred %@ %@", aStream, [[aStream streamError] copy]);
1435                /// TODO specify error better!
1436                [self _failWithError:aStream.streamError];
1437                _readBufferOffset = 0;
1438                [_readBuffer setLength:0];
1439                break;
1440
1441            }
1442
1443            case NSStreamEventEndEncountered: {
1444                [self _pumpScanner];
1445                SRFastLog(@"NSStreamEventEndEncountered %@", aStream);
1446                if (aStream.streamError) {
1447                    [self _failWithError:aStream.streamError];
1448                } else {
1449                    if (self.readyState != SR_CLOSED) {
1450                        self.readyState = SR_CLOSED;
1451                        _selfRetain = nil;
1452                    }
1453
1454                    if (!_sentClose && !_failed) {
1455                        _sentClose = YES;
1456                        // If we get closed in this state it's probably not clean because we should be sending this when we send messages
1457                        [self _performDelegateBlock:^{
1458                            if ([self.delegate respondsToSelector:@selector(webSocket:didCloseWithCode:reason:wasClean:)]) {
1459                                [self.delegate webSocket:self didCloseWithCode:SRStatusCodeGoingAway reason:@"Stream end encountered" wasClean:NO];
1460                            }
1461                        }];
1462                    }
1463                }
1464
1465                break;
1466            }
1467
1468            case NSStreamEventHasBytesAvailable: {
1469                SRFastLog(@"NSStreamEventHasBytesAvailable %@", aStream);
1470                const int bufferSize = 2048;
1471                uint8_t buffer[bufferSize];
1472
1473                while (_inputStream.hasBytesAvailable) {
1474                    NSInteger bytes_read = [_inputStream read:buffer maxLength:bufferSize];
1475
1476                    if (bytes_read > 0) {
1477                        [_readBuffer appendBytes:buffer length:bytes_read];
1478                    } else if (bytes_read < 0) {
1479                        [self _failWithError:_inputStream.streamError];
1480                    }
1481
1482                    if (bytes_read != bufferSize) {
1483                        break;
1484                    }
1485                };
1486                [self _pumpScanner];
1487                break;
1488            }
1489
1490            case NSStreamEventHasSpaceAvailable: {
1491                SRFastLog(@"NSStreamEventHasSpaceAvailable %@", aStream);
1492                [self _pumpWriting];
1493                break;
1494            }
1495
1496            default:
1497                SRFastLog(@"(default)  %@", aStream);
1498                break;
1499        }
1500    });
1501}
1502
1503@end
1504
1505
1506@implementation SRIOConsumer
1507
1508@synthesize bytesNeeded = _bytesNeeded;
1509@synthesize consumer = _scanner;
1510@synthesize handler = _handler;
1511@synthesize readToCurrentFrame = _readToCurrentFrame;
1512@synthesize unmaskBytes = _unmaskBytes;
1513
1514- (void)setupWithScanner:(stream_scanner)scanner handler:(data_callback)handler bytesNeeded:(size_t)bytesNeeded readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes;
1515{
1516    _scanner = [scanner copy];
1517    _handler = [handler copy];
1518    _bytesNeeded = bytesNeeded;
1519    _readToCurrentFrame = readToCurrentFrame;
1520    _unmaskBytes = unmaskBytes;
1521    assert(_scanner || _bytesNeeded);
1522}
1523
1524
1525@end
1526
1527
1528@implementation SRIOConsumerPool {
1529    NSUInteger _poolSize;
1530    NSMutableArray *_bufferedConsumers;
1531}
1532
1533- (id)initWithBufferCapacity:(NSUInteger)poolSize;
1534{
1535    self = [super init];
1536    if (self) {
1537        _poolSize = poolSize;
1538        _bufferedConsumers = [[NSMutableArray alloc] initWithCapacity:poolSize];
1539    }
1540    return self;
1541}
1542
1543- (id)init
1544{
1545    return [self initWithBufferCapacity:8];
1546}
1547
1548- (SRIOConsumer *)consumerWithScanner:(stream_scanner)scanner handler:(data_callback)handler bytesNeeded:(size_t)bytesNeeded readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes;
1549{
1550    SRIOConsumer *consumer = nil;
1551    if (_bufferedConsumers.count) {
1552        consumer = [_bufferedConsumers lastObject];
1553        [_bufferedConsumers removeLastObject];
1554    } else {
1555        consumer = [[SRIOConsumer alloc] init];
1556    }
1557
1558    [consumer setupWithScanner:scanner handler:handler bytesNeeded:bytesNeeded readToCurrentFrame:readToCurrentFrame unmaskBytes:unmaskBytes];
1559
1560    return consumer;
1561}
1562
1563- (void)returnConsumer:(SRIOConsumer *)consumer;
1564{
1565    if (_bufferedConsumers.count < _poolSize) {
1566        [_bufferedConsumers addObject:consumer];
1567    }
1568}
1569
1570@end
1571
1572
1573@implementation  NSURLRequest (CertificateAdditions)
1574
1575- (NSArray *)SR_SSLPinnedCertificates;
1576{
1577    return [NSURLProtocol propertyForKey:@"SR_SSLPinnedCertificates" inRequest:self];
1578}
1579
1580@end
1581
1582@implementation  NSMutableURLRequest (CertificateAdditions)
1583
1584- (NSArray *)SR_SSLPinnedCertificates;
1585{
1586    return [NSURLProtocol propertyForKey:@"SR_SSLPinnedCertificates" inRequest:self];
1587}
1588
1589- (void)setSR_SSLPinnedCertificates:(NSArray *)SR_SSLPinnedCertificates;
1590{
1591    [NSURLProtocol setProperty:SR_SSLPinnedCertificates forKey:@"SR_SSLPinnedCertificates" inRequest:self];
1592}
1593
1594@end
1595
1596@implementation NSURL (SRWebSocket)
1597
1598- (NSString *)SR_origin;
1599{
1600    NSString *scheme = [self.scheme lowercaseString];
1601
1602    if ([scheme isEqualToString:@"wss"]) {
1603        scheme = @"https";
1604    } else if ([scheme isEqualToString:@"ws"]) {
1605        scheme = @"http";
1606    }
1607
1608    if (self.port) {
1609        return [NSString stringWithFormat:@"%@://%@:%@/", scheme, self.host, self.port];
1610    } else {
1611        return [NSString stringWithFormat:@"%@://%@/", scheme, self.host];
1612    }
1613}
1614
1615@end
1616
1617//#define SR_ENABLE_LOG
1618
1619static inline void SRFastLog(NSString *format, ...)  {
1620#ifdef SR_ENABLE_LOG
1621    __block va_list arg_list;
1622    va_start (arg_list, format);
1623
1624    NSString *formattedString = [[NSString alloc] initWithFormat:format arguments:arg_list];
1625
1626    va_end(arg_list);
1627
1628    NSLog(@"[SR] %@", formattedString);
1629#endif
1630}
1631
1632
1633#ifdef HAS_ICU
1634
1635static inline int32_t validate_dispatch_data_partial_string(NSData *data) {
1636    if ([data length] > INT32_MAX) {
1637        // INT32_MAX is the limit so long as this Framework is using 32 bit ints everywhere.
1638        return -1;
1639    }
1640
1641    int32_t size = (int32_t)[data length];
1642
1643    const void * contents = [data bytes];
1644    const uint8_t *str = (const uint8_t *)contents;
1645
1646    UChar32 codepoint = 1;
1647    int32_t offset = 0;
1648    int32_t lastOffset = 0;
1649    while(offset < size && codepoint > 0)  {
1650        lastOffset = offset;
1651        U8_NEXT(str, offset, size, codepoint);
1652    }
1653
1654    if (codepoint == -1) {
1655        // Check to see if the last byte is valid or whether it was just continuing
1656        if (!U8_IS_LEAD(str[lastOffset]) || U8_COUNT_TRAIL_BYTES(str[lastOffset]) + lastOffset < (int32_t)size) {
1657
1658            size = -1;
1659        } else {
1660            uint8_t leadByte = str[lastOffset];
1661            U8_MASK_LEAD_BYTE(leadByte, U8_COUNT_TRAIL_BYTES(leadByte));
1662
1663            for (int i = lastOffset + 1; i < offset; i++) {
1664                if (U8_IS_SINGLE(str[i]) || U8_IS_LEAD(str[i]) || !U8_IS_TRAIL(str[i])) {
1665                    size = -1;
1666                }
1667            }
1668
1669            if (size != -1) {
1670                size = lastOffset;
1671            }
1672        }
1673    }
1674
1675    if (size != -1 && ![[NSString alloc] initWithBytesNoCopy:(char *)[data bytes] length:size encoding:NSUTF8StringEncoding freeWhenDone:NO]) {
1676        size = -1;
1677    }
1678
1679    return size;
1680}
1681
1682#else
1683
1684// This is a hack, and probably not optimal
1685static inline int32_t validate_dispatch_data_partial_string(NSData *data) {
1686    static const int maxCodepointSize = 3;
1687
1688    for (int i = 0; i < maxCodepointSize; i++) {
1689        NSString *str = [[NSString alloc] initWithBytesNoCopy:(char *)data.bytes length:data.length - i encoding:NSUTF8StringEncoding freeWhenDone:NO];
1690        if (str) {
1691            return data.length - i;
1692        }
1693    }
1694
1695    return -1;
1696}
1697
1698#endif
1699
1700static _SRRunLoopThread *networkThread = nil;
1701static NSRunLoop *networkRunLoop = nil;
1702
1703@implementation NSRunLoop (SRWebSocket)
1704
1705+ (NSRunLoop *)SR_networkRunLoop {
1706    static dispatch_once_t onceToken;
1707    dispatch_once(&onceToken, ^{
1708        networkThread = [[_SRRunLoopThread alloc] init];
1709        networkThread.name = @"com.squareup.SocketRocket.NetworkThread";
1710        [networkThread start];
1711        networkRunLoop = networkThread.runLoop;
1712    });
1713
1714    return networkRunLoop;
1715}
1716
1717@end
1718
1719
1720@implementation _SRRunLoopThread {
1721    dispatch_group_t _waitGroup;
1722}
1723
1724@synthesize runLoop = _runLoop;
1725
1726- (void)dealloc
1727{
1728    sr_dispatch_release(_waitGroup);
1729}
1730
1731- (id)init
1732{
1733    self = [super init];
1734    if (self) {
1735        _waitGroup = dispatch_group_create();
1736        dispatch_group_enter(_waitGroup);
1737    }
1738    return self;
1739}
1740
1741- (void)main;
1742{
1743    @autoreleasepool {
1744        _runLoop = [NSRunLoop currentRunLoop];
1745        dispatch_group_leave(_waitGroup);
1746
1747        NSTimer *timer = [[NSTimer alloc] initWithFireDate:[NSDate distantFuture] interval:0.0 target:nil selector:nil userInfo:nil repeats:NO];
1748        [_runLoop addTimer:timer forMode:NSDefaultRunLoopMode];
1749
1750        while ([_runLoop runMode:NSDefaultRunLoopMode beforeDate:[NSDate distantFuture]]) {
1751
1752        }
1753        assert(NO);
1754    }
1755}
1756
1757- (NSRunLoop *)runLoop;
1758{
1759    dispatch_group_wait(_waitGroup, DISPATCH_TIME_FOREVER);
1760    return _runLoop;
1761}
1762
1763@end
1764