--- /dev/null
+/*
+ * Copyright 2017 Google
+ *
+ * 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 "FIRMessagingSecureSocket.h"
+
+#import "GPBMessage.h"
+#import "GPBCodedOutputStream.h"
+#import "GPBUtilities.h"
+
+#import "FIRMessagingCodedInputStream.h"
+#import "FIRMessagingDefines.h"
+#import "FIRMessagingLogger.h"
+#import "FIRMessagingPacketQueue.h"
+
+static const NSUInteger kMaxBufferLength = 1024 * 1024; // 1M
+static const NSUInteger kBufferLengthIncrement = 16 * 1024; // 16k
+static const uint8_t kVersion = 40;
+static const uint8_t kInvalidTag = -1;
+
+typedef NS_ENUM(NSUInteger, FIRMessagingSecureSocketReadResult) {
+ kFIRMessagingSecureSocketReadResultNone,
+ kFIRMessagingSecureSocketReadResultIncomplete,
+ kFIRMessagingSecureSocketReadResultCorrupt,
+ kFIRMessagingSecureSocketReadResultSuccess
+};
+
+static int32_t LogicalRightShift32(int32_t value, int32_t spaces) {
+ return (int32_t)((uint32_t)(value) >> spaces);
+}
+
+static NSUInteger SerializedSize(int32_t value) {
+ NSUInteger bytes = 0;
+ while (YES) {
+ if ((value & ~0x7F) == 0) {
+ bytes += sizeof(uint8_t);
+ return bytes;
+ } else {
+ bytes += sizeof(uint8_t);
+ value = LogicalRightShift32(value, 7);
+ }
+ }
+}
+
+@interface FIRMessagingSecureSocket() <NSStreamDelegate>
+
+@property(nonatomic, readwrite, assign) FIRMessagingSecureSocketState state;
+@property(nonatomic, readwrite, strong) NSInputStream *inStream;
+@property(nonatomic, readwrite, strong) NSOutputStream *outStream;
+
+@property(nonatomic, readwrite, strong) NSMutableData *inputBuffer;
+@property(nonatomic, readwrite, assign) NSUInteger inputBufferLength;
+@property(nonatomic, readwrite, strong) NSMutableData *outputBuffer;
+@property(nonatomic, readwrite, assign) NSUInteger outputBufferLength;
+
+@property(nonatomic, readwrite, strong) FIRMessagingPacketQueue *packetQueue;
+@property(nonatomic, readwrite, assign) BOOL isVersionSent;
+@property(nonatomic, readwrite, assign) BOOL isVersionReceived;
+@property(nonatomic, readwrite, assign) BOOL isInStreamOpen;
+@property(nonatomic, readwrite, assign) BOOL isOutStreamOpen;
+
+@property(nonatomic, readwrite, strong) NSRunLoop *runLoop;
+@property(nonatomic, readwrite, strong) NSString *currentRmqIdBeingSent;
+@property(nonatomic, readwrite, assign) int8_t currentProtoTypeBeingSent;
+
+@end
+
+@implementation FIRMessagingSecureSocket
+
+- (instancetype)init {
+ self = [super init];
+ if (self) {
+ _state = kFIRMessagingSecureSocketNotOpen;
+ _inputBuffer = [NSMutableData dataWithLength:kBufferLengthIncrement];
+ _packetQueue = [[FIRMessagingPacketQueue alloc] init];
+ _currentProtoTypeBeingSent = kInvalidTag;
+ }
+ return self;
+}
+
+- (void)dealloc {
+ [self disconnect];
+}
+
+- (void)connectToHost:(NSString *)host
+ port:(NSUInteger)port
+ onRunLoop:(NSRunLoop *)runLoop {
+ _FIRMessagingDevAssert(host != nil, @"Invalid host");
+ _FIRMessagingDevAssert(runLoop != nil, @"Invalid runloop");
+ _FIRMessagingDevAssert(self.state == kFIRMessagingSecureSocketNotOpen, @"Socket is already connected");
+
+ if (!host || self.state != kFIRMessagingSecureSocketNotOpen) {
+ return;
+ }
+
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket000,
+ @"Opening secure socket to FIRMessaging service");
+ self.state = kFIRMessagingSecureSocketOpening;
+ self.runLoop = runLoop;
+ CFReadStreamRef inputStreamRef;
+ CFWriteStreamRef outputStreamRef;
+ CFStreamCreatePairWithSocketToHost(NULL,
+ (__bridge CFStringRef)host,
+ (int)port,
+ &inputStreamRef,
+ &outputStreamRef);
+ self.inStream = CFBridgingRelease(inputStreamRef);
+ self.outStream = CFBridgingRelease(outputStreamRef);
+ if (!self.inStream || !self.outStream) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket001,
+ @"Failed to initialize socket.");
+ return;
+ }
+
+ self.isInStreamOpen = NO;
+ self.isOutStreamOpen = NO;
+
+ BOOL isVOIPSocket = NO;
+
+ [self openStream:self.outStream isVOIPStream:isVOIPSocket];
+ [self openStream:self.inStream isVOIPStream:isVOIPSocket];
+}
+
+- (void)disconnect {
+ if (self.state == kFIRMessagingSecureSocketClosing) {
+ return;
+ }
+ if (!self.inStream && !self.outStream) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket002,
+ @"The socket is not open or already closed.");
+ _FIRMessagingDevAssert(self.state == kFIRMessagingSecureSocketClosed || self.state == kFIRMessagingSecureSocketNotOpen,
+ @"Socket is already disconnected.");
+ return;
+ }
+
+ self.state = kFIRMessagingSecureSocketClosing;
+ if (self.inStream) {
+ [self closeStream:self.inStream];
+ self.inStream = nil;
+ }
+ if (self.outStream) {
+ [self closeStream:self.outStream];
+ self.outStream = nil;
+ }
+ self.state = kFIRMessagingSecureSocketClosed;
+ [self.delegate didDisconnectWithSecureSocket:self];
+}
+
+- (void)sendData:(NSData *)data withTag:(int8_t)tag rmqId:(NSString *)rmqId {
+ [self.packetQueue push:[FIRMessagingPacket packetWithTag:tag rmqId:rmqId data:data]];
+ if ([self.outStream hasSpaceAvailable]) {
+ [self performWrite];
+ }
+}
+
+#pragma mark - NSStreamDelegate
+
+- (void)stream:(NSStream *)stream handleEvent:(NSStreamEvent)eventCode {
+ switch (eventCode) {
+ case NSStreamEventHasBytesAvailable:
+ if (self.state != kFIRMessagingSecureSocketOpen) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket003,
+ @"Try to read from socket that is not opened");
+ return;
+ }
+ _FIRMessagingDevAssert(stream == self.inStream, @"Incorrect stream");
+ if (![self performRead]) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket004,
+ @"Error occured when reading incoming stream");
+ [self disconnect];
+ }
+ break;
+ case NSStreamEventEndEncountered:
+ FIRMessagingLoggerDebug(
+ kFIRMessagingMessageCodeSecureSocket005, @"%@ end encountered",
+ stream == self.inStream
+ ? @"Input stream"
+ : (stream == self.outStream ? @"Output stream" : @"Unknown stream"));
+ [self disconnect];
+ break;
+ case NSStreamEventOpenCompleted:
+ if (stream == self.inStream) {
+ self.isInStreamOpen = YES;
+ } else if (stream == self.outStream) {
+ self.isOutStreamOpen = YES;
+ }
+ if (self.isInStreamOpen && self.isOutStreamOpen) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket006,
+ @"Secure socket to FIRMessaging service opened");
+ self.state = kFIRMessagingSecureSocketOpen;
+ [self.delegate secureSocketDidConnect:self];
+ }
+ break;
+ case NSStreamEventErrorOccurred: {
+ FIRMessagingLoggerDebug(
+ kFIRMessagingMessageCodeSecureSocket007, @"%@ error occurred",
+ stream == self.inStream
+ ? @"Input stream"
+ : (stream == self.outStream ? @"Output stream" : @"Unknown stream"));
+ [self disconnect];
+ break;
+ }
+ case NSStreamEventHasSpaceAvailable:
+ if (self.state != kFIRMessagingSecureSocketOpen) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket008,
+ @"Try to write to socket that is not opened");
+ return;
+ }
+ _FIRMessagingDevAssert(stream == self.outStream, @"Incorrect stream");
+ [self performWrite];
+ break;
+ default:
+ break;
+ }
+}
+
+#pragma mark - Private
+
+- (void)openStream:(NSStream *)stream isVOIPStream:(BOOL)isVOIPStream {
+ _FIRMessagingDevAssert(stream != nil, @"Invalid stream");
+ _FIRMessagingDevAssert(self.runLoop != nil, @"Invalid runloop");
+
+ if (stream) {
+ _FIRMessagingDevAssert([stream streamStatus] == NSStreamStatusNotOpen, @"Stream already open");
+ if ([stream streamStatus] != NSStreamStatusNotOpen) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket009,
+ @"stream should not be open.");
+ return;
+ }
+ [stream setProperty:NSStreamSocketSecurityLevelNegotiatedSSL
+ forKey:NSStreamSocketSecurityLevelKey];
+ if (isVOIPStream) {
+ [stream setProperty:NSStreamNetworkServiceTypeVoIP
+ forKey:NSStreamNetworkServiceType];
+ }
+ stream.delegate = self;
+ [stream scheduleInRunLoop:self.runLoop forMode:NSDefaultRunLoopMode];
+ [stream open];
+ }
+}
+
+- (void)closeStream:(NSStream *)stream {
+ _FIRMessagingDevAssert(stream != nil, @"Invalid stream");
+ _FIRMessagingDevAssert(self.runLoop != nil, @"Invalid runloop");
+
+ if (stream) {
+ [stream close];
+ [stream removeFromRunLoop:self.runLoop forMode:NSDefaultRunLoopMode];
+ stream.delegate = nil;
+ }
+}
+
+- (BOOL)performRead {
+ _FIRMessagingDevAssert(self.state == kFIRMessagingSecureSocketOpen, @"Socket should be open");
+
+ if (!self.isVersionReceived) {
+ self.isVersionReceived = YES;
+ uint8_t versionByte = 0;
+ NSInteger bytesRead = [self.inStream read:&versionByte maxLength:sizeof(uint8_t)];
+ if (bytesRead != sizeof(uint8_t) || kVersion != versionByte) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket010,
+ @"Version do not match. Received %d, Expecting %d", versionByte,
+ kVersion);
+ return NO;
+ }
+ }
+
+ while (YES) {
+ BOOL isInputBufferValid = [self.inputBuffer length] > 0;
+ _FIRMessagingDevAssert(isInputBufferValid,
+ @"Invalid input buffer size %lu. Used bytes length %lu, buffer content: %@",
+ _FIRMessaging_UL([self.inputBuffer length]),
+ _FIRMessaging_UL(self.inputBufferLength),
+ self.inputBuffer);
+ if (!isInputBufferValid) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket011,
+ @"Input buffer is not valid.");
+ return NO;
+ }
+
+ if (![self.inStream hasBytesAvailable]) {
+ break;
+ }
+
+ // try to read more data
+ uint8_t *unusedBufferPtr = (uint8_t *)self.inputBuffer.mutableBytes + self.inputBufferLength;
+ NSUInteger unusedBufferLength = [self.inputBuffer length] - self.inputBufferLength;
+ NSInteger bytesRead = [self.inStream read:unusedBufferPtr maxLength:unusedBufferLength];
+ if (bytesRead <= 0) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket012,
+ @"Failed to read input stream. Bytes read %ld, Used buffer size %lu, "
+ @"Unused buffer size %lu",
+ _FIRMessaging_UL(bytesRead), _FIRMessaging_UL(self.inputBufferLength),
+ _FIRMessaging_UL(unusedBufferLength));
+ break;
+ }
+ // did successfully read some more data
+ self.inputBufferLength += (NSUInteger)bytesRead;
+
+ if ([self.inputBuffer length] <= self.inputBufferLength) {
+ // shouldn't be reading more than 1MB of data in one go
+ if ([self.inputBuffer length] + kBufferLengthIncrement > kMaxBufferLength) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket013,
+ @"Input buffer exceed 1M, disconnect socket");
+ return NO;
+ }
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket014,
+ @"Input buffer limit exceeded. Used input buffer size %lu, "
+ @"Total input buffer size %lu. No unused buffer left. "
+ @"Increase buffer size.",
+ _FIRMessaging_UL(self.inputBufferLength),
+ _FIRMessaging_UL([self.inputBuffer length]));
+ [self.inputBuffer increaseLengthBy:kBufferLengthIncrement];
+ _FIRMessagingDevAssert([self.inputBuffer length] > self.inputBufferLength, @"Invalid buffer size");
+ }
+
+ while (self.inputBufferLength > 0 && [self.inputBuffer length] > 0) {
+ _FIRMessagingDevAssert([self.inputBuffer length] >= self.inputBufferLength,
+ @"Buffer longer than length");
+ NSRange inputRange = NSMakeRange(0, self.inputBufferLength);
+ size_t protoBytes = 0;
+ // read the actual proto data coming in
+ FIRMessagingSecureSocketReadResult readResult =
+ [self processCurrentInputBuffer:[self.inputBuffer subdataWithRange:inputRange]
+ outOffset:&protoBytes];
+ // Corrupt data encountered, stop processing.
+ if (readResult == kFIRMessagingSecureSocketReadResultCorrupt) {
+ return NO;
+ // Incomplete data, keep trying to read by loading more from the stream.
+ } else if (readResult == kFIRMessagingSecureSocketReadResultIncomplete) {
+ break;
+ }
+ _FIRMessagingDevAssert(self.inputBufferLength >= protoBytes, @"More bytes than buffer can handle");
+ // we have read (0, protoBytes) of data in the inputBuffer
+ if (protoBytes == self.inputBufferLength) {
+ // did completely read the buffer data can be reset for further processing
+ self.inputBufferLength = 0;
+ } else {
+ // delete processed bytes while maintaining the buffer size.
+ NSUInteger prevLength __unused = [self.inputBuffer length];
+ // delete the processed bytes
+ [self.inputBuffer replaceBytesInRange:NSMakeRange(0, protoBytes) withBytes:NULL length:0];
+ // reallocate more data
+ [self.inputBuffer increaseLengthBy:protoBytes];
+ _FIRMessagingDevAssert([self.inputBuffer length] == prevLength,
+ @"Invalid input buffer size %lu. Used bytes length %lu, "
+ @"buffer content: %@",
+ _FIRMessaging_UL([self.inputBuffer length]),
+ _FIRMessaging_UL(self.inputBufferLength),
+ self.inputBuffer);
+ self.inputBufferLength -= protoBytes;
+ }
+ }
+ }
+ return YES;
+}
+
+- (FIRMessagingSecureSocketReadResult)processCurrentInputBuffer:(NSData *)readData
+ outOffset:(size_t *)outOffset {
+ *outOffset = 0;
+
+ FIRMessagingCodedInputStream *input = [[FIRMessagingCodedInputStream alloc] initWithData:readData];
+ int8_t rawTag;
+ if (![input readTag:&rawTag]) {
+ return kFIRMessagingSecureSocketReadResultIncomplete;
+ }
+ int32_t length;
+ if (![input readLength:&length]) {
+ return kFIRMessagingSecureSocketReadResultIncomplete;
+ }
+ // NOTE tag can be zero for |HeartbeatPing|, and length can be zero for |Close| proto
+ _FIRMessagingDevAssert(rawTag >= 0 && length >= 0, @"Invalid tag or length");
+ if (rawTag < 0 || length < 0) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket015, @"Buffer data corrupted.");
+ return kFIRMessagingSecureSocketReadResultCorrupt;
+ }
+ NSData *data = [input readDataWithLength:(uint32_t)length];
+ if (data == nil) {
+ FIRMessagingLoggerDebug(kFIRMessagingMessageCodeSecureSocket016,
+ @"Incomplete data, buffered data length %ld, expected length %d",
+ _FIRMessaging_UL(self.inputBufferLength), length);
+ return kFIRMessagingSecureSocketReadResultIncomplete;
+ }
+ [self.delegate secureSocket:self didReceiveData:data withTag:rawTag];
+ *outOffset = input.offset;
+ return kFIRMessagingSecureSocketReadResultSuccess;
+}
+
+- (void)performWrite {
+ _FIRMessagingDevAssert(self.state == kFIRMessagingSecureSocketOpen, @"Invalid socket state");
+
+ if (!self.isVersionSent) {
+ self.isVersionSent = YES;
+ uint8_t versionByte = kVersion;
+ [self.outStream write:&versionByte maxLength:sizeof(uint8_t)];
+ }
+
+ while (!self.packetQueue.isEmpty && self.outStream.hasSpaceAvailable) {
+ if (self.outputBuffer.length == 0) {
+ // serialize new packets only when the output buffer is flushed.
+ FIRMessagingPacket *packet = [self.packetQueue pop];
+ self.currentRmqIdBeingSent = packet.rmqId;
+ self.currentProtoTypeBeingSent = packet.tag;
+ NSUInteger length = SerializedSize(packet.tag) +
+ SerializedSize((int)packet.data.length) + packet.data.length;
+ self.outputBuffer = [NSMutableData dataWithLength:length];
+ GPBCodedOutputStream *output = [GPBCodedOutputStream streamWithData:self.outputBuffer];
+ [output writeRawVarint32:packet.tag];
+ [output writeBytesNoTag:packet.data];
+ self.outputBufferLength = 0;
+ }
+
+ // flush the output buffer.
+ NSInteger written = [self.outStream write:self.outputBuffer.bytes + self.outputBufferLength
+ maxLength:self.outputBuffer.length - self.outputBufferLength];
+ if (written <= 0) {
+ continue;
+ }
+ self.outputBufferLength += (NSUInteger)written;
+ if (self.outputBufferLength >= self.outputBuffer.length) {
+ self.outputBufferLength = 0;
+ self.outputBuffer = nil;
+ [self.delegate secureSocket:self
+ didSendProtoWithTag:self.currentProtoTypeBeingSent
+ rmqId:self.currentRmqIdBeingSent];
+ self.currentRmqIdBeingSent = nil;
+ self.currentProtoTypeBeingSent = kInvalidTag;
+ }
+ }
+}
+
+@end
fnp / wl-app.git / blobdiff