Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings
This repository was archived by the owner on Jun 10, 2022. It is now read-only.
/kafka-phpPublic archive

Commitc7ab268

Browse files
authored
Merge pull request#280 from kimmelsg/bypass-ssl-authentication
Added option to disable SSL mutual authentication.
2 parentsa8f5b01 +d94a900 commitc7ab268

File tree

7 files changed

+99
-28
lines changed

7 files changed

+99
-28
lines changed

‎docs/Configure.md‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,13 +4,14 @@ Kafka-php Configuration
44
| Property| C/P| Range| Default| Desc|
55
| --| --| --| --| --|
66
| brokerVersion| C/P| 0.8.0| 0.10.1.0| User supplied broker version|
7-
| clientId| C/P|| kafka-php| This is a user supplied identifier for the client application|
7+
| clientId| C/P|| kafka-php| This is a user supplied identifier for the client application|
88
| messageMaxBytes| C/P| 1000 .. 1000000000| 1000000| Maximum transmit message size.|
99
| metadataBrokerList| C/P||| Kafka Broker server list|
1010
| metadataMaxAgeMs| C/P| 1 .. 86400000| -1| Metadata cache max age. Defaults to metadata.refresh.interval.ms * 3|
1111
| metadataRefreshIntervalMs| C/P| 10 .. 3600000| 300000| Topic metadata refresh interval in milliseconds. The metadata is automatically refreshed on error and connect. Use -1 to disable the intervalled refresh.|
1212
| metadataRequestTimeoutMs| C/P| 10 .. 900000| 60000| Non-topic request timeout in milliseconds. This is for metadata requests, etc.|
1313
| sslEnable| C/P| true/false| false| Whether enable ssl connect or not|
14+
| sslEnableAuthentication| C/P| true/false| true| Enable authentication via SSL|
1415
| sslCafile| C/P||| Location of Certificate Authority file on local filesystem which should be used with the verify_peer context option to authenticate the identity of the remote peer.|
1516
| sslLocalCert| C/P| File path|| Path to local certificate file on filesystem.|
1617
| sslLocalPk| C/P| File path|| Path to local private key file on filesystem in case of separate files for certificate (local_cert) and private key.|
@@ -23,7 +24,7 @@ Kafka-php Configuration
2324
| maxWaitTime| C|| 100| Maximum time in ms to wait for the response|
2425
| sessionTimeout| C| 1 .. 3600000| 30000| Client group session and failure detection timeout.|
2526
| rebalanceTimeout| C| 1 .. 3600000| 30000| rebalance join wait timeout|
26-
| topics| C||| Want consumer topics|
27+
| topics| C||| Want consumer topics|
2728
| isAsyn| P| true, false| false| Whether to use asynchronous production messages|
2829
| produceInterval| P| 1 .. 900000| 100| The time interval at which requests for production messages are executed when the message is produced asynchronously|
2930
| requestTimeout| P| 1 .. 900000| 6000| The total timeout of the production message, which must be greater than the timeout config parameter|

‎docs/ch/Configure.md‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,16 @@ Kafka-php 配置参数
22
==================
33

44
| Property| C/P| Range| Default| Desc|
5-
| ----| ----| ----| ----| ----|
6-
| clientId| C/P|| kafka-php| 客户端标识|
5+
| ----| ----| ----| ----| ----|
6+
| clientId| C/P|| kafka-php| 客户端标识|
77
| brokerVersion| C/P| 大于 0.8.0| 0.10.1.0| 为了计算 Kafka 请求的协议版本|
88
| metadataBrokerList| C/P||| 指定 Kafka Broker 列表，多个用逗号分隔|
99
| messageMaxBytes| C/P| 1000 .. 1000000000| 1000000| 消息最大长度|
1010
| metadataRequestTimeoutMs| C/P| 10 .. 900000| 60000| 获取 meta 信息超时时间|
1111
| metadataRefreshIntervalMs| C/P| 10 .. 3600000| 300000| 获取同步 meta 信息的时间间隔|
1212
| metadataMaxAgeMs | C/P | 1 .. 86400000 | -1 | meta 信息有效期
1313
| sslEnable| C/P| true/false| false| 是否开启 Ssl 连接|
14+
| sslEnableAuthentication| C/P| true/false| true| 通过SSL启用身份验证|
1415
| sslLocalCert| C/P| File path|| 本地证书路径|
1516
| sslLocalPk| C/P| File path|| 如果使用独立的文件来存储证书（local_cert）和私钥， 那么使用此选项来指明私钥文件的路径。|
1617
| sslVerifyPeer| C/P| true/false| false| 是否需要验证 SSL 证书。|
@@ -20,7 +21,7 @@ Kafka-php 配置参数
2021
| groupId| C||| 消费模块的分组 ID|
2122
| sessionTimeout| C| 1 .. 3600000| 30000| 分组中消费者的有效时间|
2223
| rebalanceTimeout| C| 1 .. 3600000| 30000| 分组 rebalance 等待 join 时间|
23-
| topics| C||| 将要消费的 kafka topic 名称|
24+
| topics| C||| 将要消费的 kafka topic 名称|
2425
| offsetReset| C| latest,earliest| latest| 如果消费 offset 失效的时候重置 offset 的策略|
2526
| maxBytes| C|| 65536| 单次 FETCH 请求对于单个分区请求的最大字节数|
2627
| maxWaitTime| C|| 100| 等待服务端响应 FETCH 请求的最大时间|
@@ -42,4 +43,3 @@ $config->setClientId('test');
4243
```
4344

4445
无论是消费模块还是生产模块，如果参数设置不符合规则时都会抛`\Kafka\Exception\Config` 异常
45-

‎src/Broker.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -173,7 +173,7 @@ public function getConnect(string $key, string $type, bool $modeSync = false): ?
173173
$port =null;
174174

175175
if (isset($this->brokers[$key])) {
176-
$hostname=$this->brokers[$key];
176+
$hostname =$this->brokers[$key];
177177

178178
[$host,$port] =explode(':',$hostname);
179179
}

‎src/CommonSocket.php‎

Lines changed: 39 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
usefunctionfeof;
88
usefunctionfread;
99
usefunctionfwrite;
10+
usefunctionis_null;
1011
usefunctionis_resource;
1112
usefunctionsprintf;
1213
usefunctionstream_context_create;
@@ -138,19 +139,7 @@ protected function createStream(): void
138139

139140
if ($this->config !==null &&$this->config->getSslEnable()) {// ssl connection
140141
$remoteSocket =sprintf('ssl://%s:%s',$this->host,$this->port);
141-
142-
$context =stream_context_create(
143-
[
144-
'ssl' => [
145-
'local_cert' =>$this->config->getSslLocalCert(),
146-
'local_pk' =>$this->config->getSslLocalPk(),
147-
'verify_peer' =>$this->config->getSslVerifyPeer(),
148-
'passphrase' =>$this->config->getSslPassphrase(),
149-
'cafile' =>$this->config->getSslCafile(),
150-
'peer_name' =>$this->config->getSslPeerName(),
151-
],
152-
]
153-
);
142+
$context =$this->createSslStreamContext();
154143
}
155144

156145
$this->stream =$this->createSocket($remoteSocket,$context,$errno,$errstr);
@@ -167,6 +156,43 @@ protected function createStream(): void
167156
}
168157
}
169158

159+
/**
160+
* Ecapsulation of stream_context_create for SSL connections
161+
*
162+
* @return resource $context
163+
*/
164+
protectedfunctioncreateSslStreamContext()
165+
{
166+
if (is_null($this->config)) {
167+
thrownewException('Cannot build SSL context without configuration');
168+
}
169+
170+
if ($this->config->getSslEnableAuthentication()) {
171+
returnstream_context_create(
172+
[
173+
'ssl' => [
174+
'local_cert' =>$this->config->getSslLocalCert(),
175+
'local_pk' =>$this->config->getSslLocalPk(),
176+
'verify_peer' =>$this->config->getSslVerifyPeer(),
177+
'passphrase' =>$this->config->getSslPassphrase(),
178+
'cafile' =>$this->config->getSslCafile(),
179+
'peer_name' =>$this->config->getSslPeerName(),
180+
],
181+
]
182+
);
183+
}
184+
185+
returnstream_context_create(
186+
[
187+
'ssl' => [
188+
'verify_peer' =>$this->config->getSslVerifyPeer(),
189+
'cafile' =>$this->config->getSslCafile(),
190+
'peer_name' =>$this->config->getSslPeerName(),
191+
],
192+
]
193+
);
194+
}
195+
170196
/**
171197
* Encapsulation of stream_socket_client
172198
*

‎src/Config.php‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
* @method string getSecurityProtocol()
2828
* @method bool getSslEnable()
2929
* @method void setSslEnable(bool $sslEnable)
30+
* @method bool getSslEnableAuthentication()
31+
* @method void setSslEnableAuthentication(bool $sslEnableAuthentication)
3032
* @method string getSslLocalCert()
3133
* @method string getSslLocalPk()
3234
* @method bool getSslVerifyPeer()
@@ -86,10 +88,11 @@ abstract class Config
8688
'metadataMaxAgeMs' => -1,
8789
'securityProtocol' =>self::SECURITY_PROTOCOL_PLAINTEXT,
8890
'sslEnable' =>false,// this config item will override, don't config it.
89-
'sslLocalCert' =>'',
90-
'sslLocalPk' =>'',
91+
'sslEnableAuthentication' =>true,
92+
'sslLocalCert' =>false,
93+
'sslLocalPk' =>false,
9194
'sslVerifyPeer' =>false,
92-
'sslPassphrase' =>'',
95+
'sslPassphrase' =>false,
9396
'sslCafile' =>'',
9497
'sslPeerName' =>'',
9598
'saslMechanism' =>self::SASL_MECHANISMS_PLAIN,

‎src/Consumer/Process.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -800,8 +800,8 @@ protected function stateConvert(int $errorCode, ?array $context = null): bool
800800
}
801801

802802
if ($errorCode === Protocol::OFFSET_OUT_OF_RANGE) {
803-
$resetOffset=$this->getConfig()->getOffsetReset();
804-
$offsets=$resetOffset ==='latest' ?$assign->getLastOffsets() :$assign->getOffsets();
803+
$resetOffset =$this->getConfig()->getOffsetReset();
804+
$offsets =$resetOffset ==='latest' ?$assign->getLastOffsets() :$assign->getOffsets();
805805

806806
[$topic,$partId] =$context;
807807

‎tests/Base/SocketTest.php‎

Lines changed: 44 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,46 @@ public function testCreateStreamFailure(): void
9393
}
9494

9595
publicfunctiontestCreateStreamSsl():void
96+
{
97+
$host ='127.0.0.1';
98+
$port =9192;
99+
$verifyPeer =false;
100+
$cafile =$this->root->url() .'/cafile';
101+
$peerName ='kafka';
102+
103+
$context =stream_context_create(
104+
[
105+
'ssl' => [
106+
'verify_peer' =>$verifyPeer,
107+
'cafile' =>$cafile,
108+
'peer_name' =>$peerName,
109+
],
110+
]
111+
);
112+
113+
$streamMock =$this->initStreamStub('ssl',$host,$port);
114+
115+
$streamMock->expects($this->once())
116+
->method('context')
117+
->with(stream_context_get_options($context));
118+
119+
$config =$this->getMockForAbstractClass(Config::class);
120+
$config->setSslEnable(true);
121+
$config->setSslEnableAuthentication(false);
122+
$config->setSslCafile($cafile);
123+
$config->setSslVerifyPeer($verifyPeer);
124+
$config->setSslPeerName($peerName);
125+
126+
$sasl =$this->createMock(SaslMechanism::class);
127+
$sasl->expects($this->once())
128+
->method('authenticate')
129+
->with($this->isInstanceOf(Socket::class));
130+
131+
$socket =$this->mockStreamSocketClient($host,$port,$config,$sasl);
132+
$socket->connect();
133+
}
134+
135+
publicfunctiontestCreateStreamSslAuthentication():void
96136
{
97137
$host ='127.0.0.1';
98138
$port =9192;
@@ -124,6 +164,7 @@ public function testCreateStreamSsl(): void
124164

125165
$config =$this->getMockForAbstractClass(Config::class);
126166
$config->setSslEnable(true);
167+
$config->setSslEnableAuthentication(true);
127168
$config->setSslLocalPk($localKey);
128169
$config->setSslLocalCert($localCert);
129170
$config->setSslCafile($cafile);
@@ -400,10 +441,10 @@ private function mockStreamSocketClient(
400441
$socket->method('createSocket')
401442
->willReturnCallback(
402443
function (string$remoteSocket,$context, ?int &$errno, ?string &$errstr) {
403-
$errno =99;
404-
$errstr ='my custom error';
444+
$errno =99;
445+
$errstr ='my custom error';
405446

406-
return @fopen($remoteSocket,'r+',false,$context);
447+
return @fopen($remoteSocket,'r+',false,$context);
407448
}
408449
);
409450

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp