Grcov report - context.cc

1

#include "source/extensions/common/wasm/context.h"

2

3

#include <algorithm>

4

#include <cctype>

5

#include <cstring>

6

#include <ctime>

7

#include <limits>

8

#include <memory>

9

#include <string>

10

11

#include "envoy/common/exception.h"

12

#include "envoy/extensions/wasm/v3/wasm.pb.validate.h"

13

#include "envoy/grpc/status.h"

14

#include "envoy/http/codes.h"

15

#include "envoy/local_info/local_info.h"

16

#include "envoy/network/filter.h"

17

#include "envoy/stats/sink.h"

18

#include "envoy/thread_local/thread_local.h"

19

20

#include "source/common/buffer/buffer_impl.h"

21

#include "source/common/common/assert.h"

22

#include "source/common/common/empty_string.h"

23

#include "source/common/common/enum_to_int.h"

24

#include "source/common/common/logger.h"

25

#include "source/common/common/safe_memcpy.h"

26

#include "source/common/http/header_map_impl.h"

27

#include "source/common/http/message_impl.h"

28

#include "source/common/http/utility.h"

29

#include "source/common/protobuf/utility.h"

30

#include "source/common/tracing/http_tracer_impl.h"

31

#include "source/extensions/common/wasm/plugin.h"

32

#include "source/extensions/common/wasm/wasm.h"

33

#include "source/extensions/filters/common/expr/context.h"

34

35

#include "absl/base/casts.h"

36

#include "absl/container/flat_hash_map.h"

37

#include "absl/container/node_hash_map.h"

38

#include "absl/strings/str_cat.h"

39

#include "absl/synchronization/mutex.h"

40

41

#if defined(__GNUC__)

42

#pragma GCC diagnostic push

43

#pragma GCC diagnostic ignored "-Woverloaded-virtual"

44

#endif

45

46

#include "eval/public/cel_value.h"

47

#include "eval/public/containers/field_access.h"

48

#include "eval/public/containers/field_backed_list_impl.h"

49

#include "eval/public/containers/field_backed_map_impl.h"

50

#include "eval/public/structs/cel_proto_wrapper.h"

51

52

#if defined(__GNUC__)

53

#pragma GCC diagnostic pop

54

#endif

55

56

#include "include/proxy-wasm/pairs_util.h"

57

#include "openssl/bytestring.h"

58

#include "openssl/hmac.h"

59

#include "openssl/sha.h"

60

61

using proxy_wasm::MetricType;

62

using proxy_wasm::Word;

63

64

namespace Envoy {

65

namespace Extensions {

66

namespace Common {

67

namespace Wasm {

68

69

namespace {

70

71

// FilterState prefix for CelState values.

72

constexpr absl::string_view CelStateKeyPrefix = "wasm.";

73

74

// Default behavior for Proxy-Wasm 0.2.* ABI is to not support StopIteration as

75

// a return value from onRequestHeaders() or onResponseHeaders() plugin

76

// callbacks.

77

constexpr bool DefaultAllowOnHeadersStopIteration = false;

78

79

using HashPolicy = envoy::config::route::v3::RouteAction::HashPolicy;

80

using CelState = Filters::Common::Expr::CelState;

81

using CelStatePrototype = Filters::Common::Expr::CelStatePrototype;

82

83

15

Http::RequestTrailerMapPtr buildRequestTrailerMapFromPairs(const Pairs& pairs) {

84

15

  auto map = Http::RequestTrailerMapImpl::create();

85

15

  for (auto& p : pairs) {

86

    // Note: because of the lack of a string_view interface for addCopy and

87

    // the lack of an interface to add an entry with an empty value and return

88

    // the entry, there is no efficient way to prevent either a double copy

89

    // of the value or a double lookup of the entry.

90

15

    map->addCopy(Http::LowerCaseString(std::string(p.first)), std::string(p.second));

91

15

92

15

  return map;

93

15

94

95

116

Http::RequestHeaderMapPtr buildRequestHeaderMapFromPairs(const Pairs& pairs) {

96

116

  auto map = Http::RequestHeaderMapImpl::create();

97

178

  for (auto& p : pairs) {

98

    // Note: because of the lack of a string_view interface for addCopy and

99

    // the lack of an interface to add an entry with an empty value and return

100

    // the entry, there is no efficient way to prevent either a double copy

101

    // of the value or a double lookup of the entry.

102

178

    map->addCopy(Http::LowerCaseString(std::string(p.first)), std::string(p.second));

103

178

104

116

  return map;

105

116

106

107

286

template <typename P> static uint32_t headerSize(const P& p) { return p ? p->size() : 0; }

108

109

} // namespace

110

111

// Test support.

112

113

1179

size_t Buffer::size() const {

114

1179

  if (const_buffer_instance_) {

115

746

    return const_buffer_instance_->length();

116

746

117

433

  return proxy_wasm::BufferBase::size();

118

1179

119

120

WasmResult Buffer::copyTo(WasmBase* wasm, size_t start, size_t length, uint64_t ptr_ptr,

121

376

                          uint64_t size_ptr) const {

122

376

  if (const_buffer_instance_) {

123

    // Validate that the requested range is within bounds before allocating.

124

190

    if (start + length > const_buffer_instance_->length()) {

125

1

      return WasmResult::InvalidMemoryAccess;

126

1

127

189

    uint64_t pointer;

128

189

    auto p = wasm->allocMemory(length, &pointer);

129

189

    if (!p) {

130

      return WasmResult::InvalidMemoryAccess;

131

132

189

    const_buffer_instance_->copyOut(start, length, p);

133

189

    if (!wasm->wasm_vm()->setWord(ptr_ptr, Word(pointer))) {

134

1

      return WasmResult::InvalidMemoryAccess;

135

1

136

188

    if (!wasm->wasm_vm()->setWord(size_ptr, Word(length))) {

137

1

      return WasmResult::InvalidMemoryAccess;

138

1

139

187

    return WasmResult::Ok;

140

188

141

186

  return proxy_wasm::BufferBase::copyTo(wasm, start, length, ptr_ptr, size_ptr);

142

376

143

144

82

WasmResult Buffer::copyFrom(size_t start, size_t length, std::string_view data) {

145

82

  if (buffer_instance_) {

146

80

    if (start == 0) {

147

54

      if (length != 0) {

148

45

        buffer_instance_->drain(length);

149

45

150

54

      buffer_instance_->prepend(toAbslStringView(data));

151

54

      return WasmResult::Ok;

152

71

    } else if (start >= buffer_instance_->length()) {

153

25

      buffer_instance_->add(toAbslStringView(data));

154

25

      return WasmResult::Ok;

155

26

    } else {

156

1

      return WasmResult::BadArgument;

157

1

158

80

159

2

  if (const_buffer_instance_) { // This buffer is immutable.

160

1

    return WasmResult::BadArgument;

161

1

162

1

  return proxy_wasm::BufferBase::copyFrom(start, length, data);

163

2

164

165

16

Context::Context() = default;

166

1051

Context::Context(Wasm* wasm) : ContextBase(wasm) {

167

1051

  if (wasm != nullptr) {

168

1051

    abi_version_ = wasm->abi_version_;

169

1051

170

1051

171

741

Context::Context(Wasm* wasm, const PluginSharedPtr& plugin) : ContextBase(wasm, plugin) {

172

741

  if (wasm != nullptr) {

173

741

    abi_version_ = wasm->abi_version_;

174

741

175

741

  root_local_info_ = &this->plugin()->localInfo();

176

741

  allow_on_headers_stop_iteration_ = PROTOBUF_GET_WRAPPED_OR_DEFAULT(

177

741

      this->plugin()->wasmConfig().config(), allow_on_headers_stop_iteration,

178

741

      DefaultAllowOnHeadersStopIteration);

179

741

180

Context::Context(Wasm* wasm, uint32_t root_context_id, PluginHandleSharedPtr plugin_handle)

181

331

    : ContextBase(wasm, root_context_id, plugin_handle), plugin_handle_(plugin_handle) {

182

331

  if (wasm != nullptr) {

183

316

    abi_version_ = wasm->abi_version_;

184

316

185

331

  allow_on_headers_stop_iteration_ = PROTOBUF_GET_WRAPPED_OR_DEFAULT(

186

331

      plugin()->wasmConfig().config(), allow_on_headers_stop_iteration,

187

331

      DefaultAllowOnHeadersStopIteration);

188

331

189

190

7118

WasmBase* Context::wasm() const { return wasm_; }

191

1509

Wasm* Context::envoyWasm() const { return static_cast<Wasm*>(wasm_); }

192

1912

Plugin* Context::plugin() const { return static_cast<Plugin*>(plugin_.get()); }

193

130

Context* Context::rootContext() const { return static_cast<Context*>(root_context()); }

194

230

Upstream::ClusterManager& Context::clusterManager() const { return envoyWasm()->clusterManager(); }

195

196

2

void Context::error(std::string_view message) { ENVOY_LOG(trace, message); }

197

198

2

uint64_t Context::getCurrentTimeNanoseconds() {

199

2

  return std::chrono::duration_cast<std::chrono::nanoseconds>(

200

2

             envoyWasm()->time_source_.systemTime().time_since_epoch())

201

2

      .count();

202

2

203

204

2

uint64_t Context::getMonotonicTimeNanoseconds() {

205

2

  return std::chrono::duration_cast<std::chrono::nanoseconds>(

206

2

             envoyWasm()->time_source_.monotonicTime().time_since_epoch())

207

2

      .count();

208

2

209

210

36

void Context::onCloseTCP() {

211

36

  if (tcp_connection_closed_ || !in_vm_context_created_) {

212

9

    return;

213

9

214

27

  tcp_connection_closed_ = true;

215

27

  onDone();

216

27

  onLog();

217

27

  onDelete();

218

27

219

220

void Context::onResolveDns(uint32_t token, Envoy::Network::DnsResolver::ResolutionStatus status,

221

7

                           std::list<Envoy::Network::DnsResponse>&& response) {

222

7

  proxy_wasm::DeferAfterCallActions actions(this);

223

7

  if (envoyWasm()->isFailed() || !envoyWasm()->on_resolve_dns_) {

224

1

    return;

225

1

226

6

  if (status != Network::DnsResolver::ResolutionStatus::Completed) {

227

2

    buffer_.set("");

228

2

    envoyWasm()->on_resolve_dns_(this, id_, token, 0);

229

2

    return;

230

2

231

  // buffer format:

232

  //    4 bytes number of entries = N

233

  //    N * 4 bytes TTL for each entry

234

  //    N * null-terminated addresses

235

4

  uint32_t s = 4; // length

236

4

  for (auto& e : response) {

237

4

    s += 4;                                                // for TTL

238

4

    s += e.addrInfo().address_->asStringView().size() + 1; // null terminated.

239

4

240

4

  auto buffer = std::unique_ptr<char[]>(new char[s]);

241

4

  char* b = buffer.get();

242

4

  uint32_t n = response.size();

243

4

  safeMemcpyUnsafeDst(b, &n);

244

4

  b += sizeof(uint32_t);

245

4

  for (auto& e : response) {

246

4

    uint32_t ttl = e.addrInfo().ttl_.count();

247

4

    safeMemcpyUnsafeDst(b, &ttl);

248

4

    b += sizeof(uint32_t);

249

4

};

250

4

  for (auto& e : response) {

251

4

    memcpy(b, e.addrInfo().address_->asStringView().data(), // NOLINT(safe-memcpy)

252

4

           e.addrInfo().address_->asStringView().size());

253

4

    b += e.addrInfo().address_->asStringView().size();

254

4

    *b++ = 0;

255

4

};

256

4

  buffer_.set(std::move(buffer), s);

257

4

  envoyWasm()->on_resolve_dns_(this, id_, token, s);

258

4

259

260

8

template <typename I> inline uint32_t align(uint32_t i) {

261

8

  return (i + sizeof(I) - 1) & ~(sizeof(I) - 1);

262

8

263

264

8

template <typename I> inline char* align(char* p) {

265

8

  return reinterpret_cast<char*>((reinterpret_cast<uintptr_t>(p) + sizeof(I) - 1) &

266

8

                                 ~(sizeof(I) - 1));

267

8

268

269

7

void Context::onStatsUpdate(Envoy::Stats::MetricSnapshot& snapshot) {

270

7

  proxy_wasm::DeferAfterCallActions actions(this);

271

7

  if (envoyWasm()->isFailed() || !envoyWasm()->on_stats_update_) {

272

1

    return;

273

1

274

  // buffer format:

275

  //  uint32 size of block of this type

276

  //  uint32 type

277

  //  uint32 count

278

  //    uint32 length of name

279

  //    name

280

  //    8 byte alignment padding

281

  //    8 bytes of absolute value

282

  //    8 bytes of delta  (if appropriate, e.g. for counters)

283

  //  uint32 size of block of this type

284

285

6

  uint32_t counter_block_size = 3 * sizeof(uint32_t); // type of stat

286

6

  uint32_t num_counters = snapshot.counters().size();

287

6

  uint32_t counter_type = 1;

288

289

6

  uint32_t gauge_block_size = 3 * sizeof(uint32_t); // type of stat

290

6

  uint32_t num_gauges = snapshot.gauges().size();

291

6

  uint32_t gauge_type = 2;

292

293

6

  uint32_t n = 0;

294

6

  uint64_t v = 0;

295

296

8

  for (const auto& counter : snapshot.counters()) {

297

4

    if (counter.counter_.get().used()) {

298

4

      counter_block_size += sizeof(uint32_t) + counter.counter_.get().name().size();

299

4

      counter_block_size = align<uint64_t>(counter_block_size + 2 * sizeof(uint64_t));

300

4

301

4

302

303

8

  for (const auto& gauge : snapshot.gauges()) {

304

4

    if (gauge.get().used()) {

305

4

      gauge_block_size += sizeof(uint32_t) + gauge.get().name().size();

306

4

      gauge_block_size += align<uint64_t>(gauge_block_size + sizeof(uint64_t));

307

4

308

4

309

310

6

  auto buffer = std::unique_ptr<char[]>(new char[counter_block_size + gauge_block_size]);

311

6

  char* b = buffer.get();

312

313

6

  safeMemcpyUnsafeDst(b, &counter_block_size);

314

6

  b += sizeof(uint32_t);

315

6

  safeMemcpyUnsafeDst(b, &counter_type);

316

6

  b += sizeof(uint32_t);

317

6

  safeMemcpyUnsafeDst(b, &num_counters);

318

6

  b += sizeof(uint32_t);

319

320

8

  for (const auto& counter : snapshot.counters()) {

321

4

    if (counter.counter_.get().used()) {

322

4

      n = counter.counter_.get().name().size();

323

4

      safeMemcpyUnsafeDst(b, &n);

324

4

      b += sizeof(uint32_t);

325

4

      memcpy(b, counter.counter_.get().name().data(), // NOLINT(safe-memcpy)

326

4

             counter.counter_.get().name().size());

327

4

      b = align<uint64_t>(b + counter.counter_.get().name().size());

328

4

      v = counter.counter_.get().value();

329

4

      safeMemcpyUnsafeDst(b, &v);

330

4

      b += sizeof(uint64_t);

331

4

      v = counter.delta_;

332

4

      safeMemcpyUnsafeDst(b, &v);

333

4

      b += sizeof(uint64_t);

334

4

335

4

336

337

6

  safeMemcpyUnsafeDst(b, &gauge_block_size);

338

6

  b += sizeof(uint32_t);

339

6

  safeMemcpyUnsafeDst(b, &gauge_type);

340

6

  b += sizeof(uint32_t);

341

6

  safeMemcpyUnsafeDst(b, &num_gauges);

342

6

  b += sizeof(uint32_t);

343

344

8

  for (const auto& gauge : snapshot.gauges()) {

345

4

    if (gauge.get().used()) {

346

4

      n = gauge.get().name().size();

347

4

      safeMemcpyUnsafeDst(b, &n);

348

4

      b += sizeof(uint32_t);

349

4

      memcpy(b, gauge.get().name().data(), gauge.get().name().size()); // NOLINT(safe-memcpy)

350

4

      b = align<uint64_t>(b + gauge.get().name().size());

351

4

      v = gauge.get().value();

352

4

      safeMemcpyUnsafeDst(b, &v);

353

4

      b += sizeof(uint64_t);

354

4

355

4

356

6

  buffer_.set(std::move(buffer), counter_block_size + gauge_block_size);

357

6

  envoyWasm()->on_stats_update_(this, id_, counter_block_size + gauge_block_size);

358

6

359

360

// Native serializer carrying over bit representation from CEL value to the extension.

361

// This implementation assumes that the value type is static and known to the consumer.

362

676

WasmResult serializeValue(Filters::Common::Expr::CelValue value, std::string* result) {

363

676

  using Filters::Common::Expr::CelValue;

364

676

  int64_t out_int64;

365

676

  uint64_t out_uint64;

366

676

  double out_double;

367

676

  bool out_bool;

368

676

  const Protobuf::Message* out_message;

369

676

  switch (value.type()) {

370

632

  case CelValue::Type::kString:

371

632

    result->assign(value.StringOrDie().value().data(), value.StringOrDie().value().size());

372

632

    return WasmResult::Ok;

373

8

  case CelValue::Type::kBytes:

374

8

    result->assign(value.BytesOrDie().value().data(), value.BytesOrDie().value().size());

375

8

    return WasmResult::Ok;

376

4

  case CelValue::Type::kInt64:

377

4

    out_int64 = value.Int64OrDie();

378

4

    result->assign(reinterpret_cast<const char*>(&out_int64), sizeof(int64_t));

379

4

    return WasmResult::Ok;

380

4

  case CelValue::Type::kUint64:

381

4

    out_uint64 = value.Uint64OrDie();

382

4

    result->assign(reinterpret_cast<const char*>(&out_uint64), sizeof(uint64_t));

383

4

    return WasmResult::Ok;

384

2

  case CelValue::Type::kDouble:

385

2

    out_double = value.DoubleOrDie();

386

2

    result->assign(reinterpret_cast<const char*>(&out_double), sizeof(double));

387

2

    return WasmResult::Ok;

388

2

  case CelValue::Type::kBool:

389

2

    out_bool = value.BoolOrDie();

390

2

    result->assign(reinterpret_cast<const char*>(&out_bool), sizeof(bool));

391

2

    return WasmResult::Ok;

392

3

  case CelValue::Type::kDuration:

393

    // Warning: loss of precision to nanoseconds

394

3

    out_int64 = absl::ToInt64Nanoseconds(value.DurationOrDie());

395

3

    result->assign(reinterpret_cast<const char*>(&out_int64), sizeof(int64_t));

396

3

    return WasmResult::Ok;

397

2

  case CelValue::Type::kTimestamp:

398

    // Warning: loss of precision to nanoseconds

399

2

    out_int64 = absl::ToUnixNanos(value.TimestampOrDie());

400

2

    result->assign(reinterpret_cast<const char*>(&out_int64), sizeof(int64_t));

401

2

    return WasmResult::Ok;

402

4

  case CelValue::Type::kMessage:

403

4

    out_message = value.MessageOrDie();

404

4

    result->clear();

405

4

    if (!out_message || out_message->SerializeToString(result)) {

406

4

      return WasmResult::Ok;

407

4

408

    return WasmResult::SerializationFailure;

409

6

  case CelValue::Type::kMap: {

410

6

    const auto& map = *value.MapOrDie();

411

6

    auto keys_list = map.ListKeys();

412

6

    if (!keys_list.ok()) {

413

1

      return WasmResult::SerializationFailure;

414

1

415

5

    const auto& keys = *keys_list.value();

416

5

    std::vector<std::pair<std::string, std::string>> pairs(map.size(), std::make_pair("", ""));

417

8

    for (auto i = 0; i < map.size(); i++) {

418

5

      if (serializeValue(keys[i], &pairs[i].first) != WasmResult::Ok) {

419

1

        return WasmResult::SerializationFailure;

420

1

421

4

      if (serializeValue(map[keys[i]].value(), &pairs[i].second) != WasmResult::Ok) {

422

1

        return WasmResult::SerializationFailure;

423

1

424

4

425

3

    auto size = proxy_wasm::PairsUtil::pairsSize(pairs);

426

    // prevent string inlining which violates byte alignment

427

3

    result->resize(std::max(size, static_cast<size_t>(30)));

428

3

    if (!proxy_wasm::PairsUtil::marshalPairs(pairs, result->data(), size)) {

429

      return WasmResult::SerializationFailure;

430

431

3

    result->resize(size);

432

3

    return WasmResult::Ok;

433

3

434

4

  case CelValue::Type::kList: {

435

4

    const auto& list = *value.ListOrDie();

436

4

    std::vector<std::pair<std::string, std::string>> pairs(list.size(), std::make_pair("", ""));

437

9

    for (auto i = 0; i < list.size(); i++) {

438

6

      if (serializeValue(list[i], &pairs[i].first) != WasmResult::Ok) {

439

1

        return WasmResult::SerializationFailure;

440

1

441

6

442

3

    auto size = proxy_wasm::PairsUtil::pairsSize(pairs);

443

    // prevent string inlining which violates byte alignment

444

3

    if (size < 30) {

445

1

      result->reserve(30);

446

1

447

3

    result->resize(size);

448

3

    if (!proxy_wasm::PairsUtil::marshalPairs(pairs, result->data(), size)) {

449

      return WasmResult::SerializationFailure;

450

451

3

    return WasmResult::Ok;

452

3

453

5

  default:

454

5

    break;

455

676

456

5

  return WasmResult::SerializationFailure;

457

676

458

459

#define PROPERTY_TOKENS(_f) _f(PLUGIN_NAME) _f(PLUGIN_ROOT_ID) _f(PLUGIN_VM_ID) _f(CONNECTION_ID)

460

461

392

static inline std::string downCase(std::string s) {

462

4900

  std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return std::tolower(c); });

463

392

  return s;

464

392

465

466

#define _DECLARE(_t) _t,

467

enum class PropertyToken { PROPERTY_TOKENS(_DECLARE) };

468

#undef _DECLARE

469

470

#define _PAIR(_t) {downCase(#_t), PropertyToken::_t},

471

static absl::flat_hash_map<std::string, PropertyToken> property_tokens = {PROPERTY_TOKENS(_PAIR)};

472

#undef _PAIR

473

474

absl::optional<google::api::expr::runtime::CelValue>

475

8

Context::FindValue(absl::string_view name, Protobuf::Arena* arena) const {

476

8

  return findValue(name, arena, false);

477

8

478

479

absl::optional<google::api::expr::runtime::CelValue>

480

739

Context::findValue(absl::string_view name, Protobuf::Arena* arena, bool last) const {

481

739

  using google::api::expr::runtime::CelProtoWrapper;

482

739

  using google::api::expr::runtime::CelValue;

483

484

739

  const StreamInfo::StreamInfo* info = getConstRequestStreamInfo();

485

  // In order to delegate to the StreamActivation method, we have to set the

486

  // context properties to match the Wasm context properties in all callbacks

487

  // (e.g. onLog or onEncodeHeaders) for the duration of the call.

488

739

  if (root_local_info_) {

489

648

    local_info_ = root_local_info_;

490

652

  } else if (plugin_) {

491

87

    local_info_ = &plugin()->localInfo();

492

87

493

739

  activation_info_ = info;

494

739

  activation_request_headers_ = request_headers_ ? request_headers_ : access_log_request_headers_;

495

739

  activation_response_headers_ =

496

739

      response_headers_ ? response_headers_ : access_log_response_headers_;

497

739

  activation_response_trailers_ =

498

739

      response_trailers_ ? response_trailers_ : access_log_response_trailers_;

499

739

  auto value = StreamActivation::FindValue(name, arena);

500

739

  resetActivation();

501

739

  if (value) {

502

56

    return value;

503

56

504

505

  // Convert into a dense token to enable a jump table implementation.

506

683

  auto part_token = property_tokens.find(name);

507

683

  if (part_token == property_tokens.end()) {

508

105

    if (info) {

509

38

      std::string key = absl::StrCat(CelStateKeyPrefix, name);

510

38

      const CelState* state = info->filterState().getDataReadOnly<CelState>(key);

511

38

      if (state == nullptr) {

512

6

        if (info->upstreamInfo().has_value() &&

513

6

            info->upstreamInfo().value().get().upstreamFilterState() != nullptr) {

514

          state =

515

              info->upstreamInfo().value().get().upstreamFilterState()->getDataReadOnly<CelState>(

516

                  key);

517

518

6

519

38

      if (state != nullptr) {

520

32

        return state->exprValue(arena, last);

521

32

522

38

523

73

    return {};

524

105

525

526

578

  switch (part_token->second) {

527

6

  case PropertyToken::CONNECTION_ID: {

528

6

    auto conn = getConnection();

529

6

    if (conn) {

530

2

      return CelValue::CreateUint64(conn->id());

531

2

532

4

    break;

533

6

534

7

  case PropertyToken::PLUGIN_NAME:

535

7

    if (plugin_) {

536

6

      return CelValue::CreateStringView(plugin()->name_);

537

6

538

1

    break;

539

560

  case PropertyToken::PLUGIN_ROOT_ID:

540

559

    return CelValue::CreateStringView(toAbslStringView(root_id()));

541

6

  case PropertyToken::PLUGIN_VM_ID:

542

6

    return CelValue::CreateStringView(toAbslStringView(envoyWasm()->vm_id()));

543

578

544

5

  return {};

545

578

546

547

731

WasmResult Context::getProperty(std::string_view path, std::string* result) {

548

731

  using google::api::expr::runtime::CelValue;

549

550

731

  bool first = true;

551

731

  CelValue value;

552

731

  Protobuf::Arena arena;

553

554

731

  size_t start = 0;

555

1530

  while (true) {

556

1530

    if (start >= path.size()) {

557

651

      break;

558

651

559

560

879

    size_t end = path.find('\0', start);

561

879

    if (end == absl::string_view::npos) {

562

382

      end = start + path.size();

563

382

564

879

    auto part = path.substr(start, end - start);

565

879

    start = end + 1;

566

567

879

    if (first) {

568

      // top-level identifier

569

731

      first = false;

570

731

      auto top_value = findValue(toAbslStringView(part), &arena, start >= path.size());

571

731

      if (!top_value.has_value()) {

572

74

        return WasmResult::NotFound;

573

74

574

657

      value = top_value.value();

575

689

    } else if (value.IsMap()) {

576

91

      auto& map = *value.MapOrDie();

577

91

      auto field = map[CelValue::CreateStringView(toAbslStringView(part))];

578

91

      if (!field.has_value()) {

579

2

        return WasmResult::NotFound;

580

2

581

89

      value = field.value();

582

89

    } else if (value.IsMessage()) {

583

51

      auto msg = value.MessageOrDie();

584

51

      if (msg == nullptr) {

585

        return WasmResult::NotFound;

586

587

51

      const Protobuf::Descriptor* desc = msg->GetDescriptor();

588

51

      const Protobuf::FieldDescriptor* field_desc = desc->FindFieldByName(std::string(part));

589

51

      if (field_desc == nullptr) {

590

        return WasmResult::NotFound;

591

592

51

      if (field_desc->is_map()) {

593

14

        value = CelValue::CreateMap(

594

14

            Protobuf::Arena::Create<google::api::expr::runtime::FieldBackedMapImpl>(

595

14

                &arena, msg, field_desc, &arena));

596

39

      } else if (field_desc->is_repeated()) {

597

2

        value = CelValue::CreateList(

598

2

            Protobuf::Arena::Create<google::api::expr::runtime::FieldBackedListImpl>(

599

2

                &arena, msg, field_desc, &arena));

600

35

      } else {

601

35

        auto status =

602

35

            google::api::expr::runtime::CreateValueFromSingleField(msg, field_desc, &arena, &value);

603

35

        if (!status.ok()) {

604

          return WasmResult::InternalFailure;

605

606

35

607

51

    } else if (value.IsList()) {

608

6

      auto& list = *value.ListOrDie();

609

6

      int idx = 0;

610

6

      if (!absl::SimpleAtoi(toAbslStringView(part), &idx)) {

611

2

        return WasmResult::NotFound;

612

2

613

4

      if (idx < 0 || idx >= list.size()) {

614

2

        return WasmResult::NotFound;

615

2

616

2

      value = list[idx];

617

2

    } else {

618

      return WasmResult::NotFound;

619

620

879

621

622

651

  return serializeValue(value, result);

623

731

624

625

// Header/Trailer/Metadata Maps.

626

297

Http::HeaderMap* Context::getMap(WasmHeaderMapType type) {

627

297

  switch (type) {

628

120

  case WasmHeaderMapType::RequestHeaders:

629

120

    return request_headers_;

630

10

  case WasmHeaderMapType::RequestTrailers:

631

10

    if (request_trailers_ == nullptr && request_body_buffer_ && end_of_stream_ &&

632

10

        decoder_callbacks_) {

633

6

      request_trailers_ = &decoder_callbacks_->addDecodedTrailers();

634

6

635

10

    return request_trailers_;

636

82

  case WasmHeaderMapType::ResponseHeaders:

637

82

    return response_headers_;

638

59

  case WasmHeaderMapType::ResponseTrailers:

639

59

    if (response_trailers_ == nullptr && response_body_buffer_ && end_of_stream_ &&

640

59

        encoder_callbacks_) {

641

3

      response_trailers_ = &encoder_callbacks_->addEncodedTrailers();

642

3

643

59

    return response_trailers_;

644

26

  default:

645

26

    return nullptr;

646

297

647

297

648

649

294

const Http::HeaderMap* Context::getConstMap(WasmHeaderMapType type) {

650

294

  switch (type) {

651

156

  case WasmHeaderMapType::RequestHeaders:

652

156

    if (access_log_phase_) {

653

18

      return access_log_request_headers_;

654

18

655

138

    return request_headers_;

656

10

  case WasmHeaderMapType::RequestTrailers:

657

10

    if (access_log_phase_) {

658

8

      return nullptr;

659

8

660

2

    return request_trailers_;

661

42

  case WasmHeaderMapType::ResponseHeaders:

662

42

    if (access_log_phase_) {

663

19

      return access_log_response_headers_;

664

19

665

23

    return response_headers_;

666

12

  case WasmHeaderMapType::ResponseTrailers:

667

12

    if (access_log_phase_) {

668

8

      return access_log_response_trailers_;

669

8

670

4

    return response_trailers_;

671

25

  case WasmHeaderMapType::GrpcReceiveInitialMetadata:

672

25

    return rootContext()->grpc_receive_initial_metadata_.get();

673

10

  case WasmHeaderMapType::GrpcReceiveTrailingMetadata:

674

10

    return rootContext()->grpc_receive_trailing_metadata_.get();

675

20

  case WasmHeaderMapType::HttpCallResponseHeaders: {

676

20

    Envoy::Http::ResponseMessagePtr* response = rootContext()->http_call_response_;

677

20

    if (response) {

678

4

      return &(*response)->headers();

679

4

680

16

    return nullptr;

681

20

682

19

  case WasmHeaderMapType::HttpCallResponseTrailers: {

683

19

    Envoy::Http::ResponseMessagePtr* response = rootContext()->http_call_response_;

684

19

    if (response) {

685

3

      return (*response)->trailers();

686

3

687

16

    return nullptr;

688

19

689

294

690

  IS_ENVOY_BUG("unexpected");

691

  return nullptr;

692

294

693

694

WasmResult Context::addHeaderMapValue(WasmHeaderMapType type, std::string_view key,

695

124

                                      std::string_view value) {

696

124

  auto map = getMap(type);

697

124

  if (!map) {

698

76

    return WasmResult::BadArgument;

699

76

700

48

  const Http::LowerCaseString lower_key{std::string(key)};

701

48

  map->addCopy(lower_key, std::string(value));

702

48

  onHeadersModified(type);

703

48

  return WasmResult::Ok;

704

124

705

706

WasmResult Context::getHeaderMapValue(WasmHeaderMapType type, std::string_view key,

707

288

                                      std::string_view* value) {

708

288

  auto map = getConstMap(type);

709

288

  if (!map) {

710

53

    if (access_log_phase_) {

711

      // Maps might point to nullptr in the access log phase.

712

19

      if (envoyWasm()->abiVersion() == proxy_wasm::AbiVersion::ProxyWasm_0_1_0) {

713

        *value = "";

714

        return WasmResult::Ok;

715

19

      } else {

716

19

        return WasmResult::NotFound;

717

19

718

19

719

    // Requested map type is not currently available.

720

34

    return WasmResult::BadArgument;

721

53

722

235

  const Http::LowerCaseString lower_key{std::string(key)};

723

235

  const auto entry = map->get(lower_key);

724

235

  if (entry.empty()) {

725

68

    if (envoyWasm()->abiVersion() == proxy_wasm::AbiVersion::ProxyWasm_0_1_0) {

726

1

      *value = "";

727

1

      return WasmResult::Ok;

728

68

    } else {

729

67

      return WasmResult::NotFound;

730

67

731

68

732

  // TODO(kyessenov, PiotrSikora): This needs to either return a concatenated list of values, or

733

  // the ABI needs to be changed to return multiple values. This is a potential security issue.

734

167

  *value = toStdStringView(entry[0]->value().getStringView());

735

167

  return WasmResult::Ok;

736

235

737

738

6

Pairs headerMapToPairs(const Http::HeaderMap* map) {

739

6

  if (!map) {

740

3

    return {};

741

3

742

3

  Pairs pairs;

743

3

  pairs.reserve(map->size());

744

3

  map->iterate([&pairs](const Http::HeaderEntry& header) -> Http::HeaderMap::Iterate {

745

3

    pairs.push_back(std::make_pair(toStdStringView(header.key().getStringView()),

746

3

                                   toStdStringView(header.value().getStringView())));

747

3

    return Http::HeaderMap::Iterate::Continue;

748

3

});

749

3

  return pairs;

750

6

751

752

6

WasmResult Context::getHeaderMapPairs(WasmHeaderMapType type, Pairs* result) {

753

6

  *result = headerMapToPairs(getConstMap(type));

754

6

  return WasmResult::Ok;

755

6

756

757

7

WasmResult Context::setHeaderMapPairs(WasmHeaderMapType type, const Pairs& pairs) {

758

7

  auto map = getMap(type);

759

7

  if (!map) {

760

4

    return WasmResult::BadArgument;

761

4

762

3

  std::vector<std::string> keys;

763

6

  map->iterate([&keys](const Http::HeaderEntry& header) -> Http::HeaderMap::Iterate {

764

6

    keys.push_back(std::string(header.key().getStringView()));

765

6

    return Http::HeaderMap::Iterate::Continue;

766

6

});

767

6

  for (auto& k : keys) {

768

6

    const Http::LowerCaseString lower_key{k};

769

6

    map->remove(lower_key);

770

6

771

3

  for (auto& p : pairs) {

772

3

    const Http::LowerCaseString lower_key{std::string(p.first)};

773

3

    map->addCopy(lower_key, std::string(p.second));

774

3

775

3

  onHeadersModified(type);

776

3

  return WasmResult::Ok;

777

7

778

779

31

WasmResult Context::removeHeaderMapValue(WasmHeaderMapType type, std::string_view key) {

780

31

  auto map = getMap(type);

781

31

  if (!map) {

782

26

    return WasmResult::BadArgument;

783

26

784

5

  const Http::LowerCaseString lower_key{std::string(key)};

785

5

  map->remove(lower_key);

786

5

  onHeadersModified(type);

787

5

  return WasmResult::Ok;

788

31

789

790

WasmResult Context::replaceHeaderMapValue(WasmHeaderMapType type, std::string_view key,

791

83

                                          std::string_view value) {

792

83

  auto map = getMap(type);

793

83

  if (!map) {

794

2

    return WasmResult::BadArgument;

795

2

796

81

  const Http::LowerCaseString lower_key{std::string(key)};

797

81

  map->setCopy(lower_key, toAbslStringView(value));

798

81

  onHeadersModified(type);

799

81

  return WasmResult::Ok;

800

83

801

802

52

WasmResult Context::getHeaderMapSize(WasmHeaderMapType type, uint32_t* result) {

803

52

  auto map = getMap(type);

804

52

  if (!map) {

805

26

    return WasmResult::BadArgument;

806

26

807

26

  *result = map->byteSize();

808

26

  return WasmResult::Ok;

809

52

810

811

// Buffer

812

813

839

BufferInterface* Context::getBuffer(WasmBufferType type) {

814

839

  Envoy::Http::ResponseMessagePtr* response = nullptr;

815

839

  switch (type) {

816

8

  case WasmBufferType::CallData:

817

    // Set before the call.

818

8

    return &buffer_;

819

173

  case WasmBufferType::VmConfiguration:

820

173

    return buffer_.set(envoyWasm()->vm_configuration());

821

42

  case WasmBufferType::PluginConfiguration:

822

42

    if (temp_plugin_) {

823

32

      return buffer_.set(temp_plugin_->plugin_configuration_);

824

32

825

10

    return nullptr;

826

371

  case WasmBufferType::HttpRequestBody:

827

371

    if (buffering_request_body_ && decoder_callbacks_) {

828

      // We need the mutable version, so capture it using a callback.

829

      // TODO: consider adding a mutableDecodingBuffer() interface.

830

118

      ::Envoy::Buffer::Instance* buffer_instance{};

831

118

      decoder_callbacks_->modifyDecodingBuffer(

832

118

          [&buffer_instance](::Envoy::Buffer::Instance& buffer) { buffer_instance = &buffer; });

833

118

      return buffer_.set(buffer_instance);

834

118

835

253

    return buffer_.set(request_body_buffer_);

836

206

  case WasmBufferType::HttpResponseBody:

837

206

    if (buffering_response_body_ && encoder_callbacks_) {

838

      // TODO: consider adding a mutableDecodingBuffer() interface.

839

76

      ::Envoy::Buffer::Instance* buffer_instance{};

840

76

      encoder_callbacks_->modifyEncodingBuffer(

841

76

          [&buffer_instance](::Envoy::Buffer::Instance& buffer) { buffer_instance = &buffer; });

842

76

      return buffer_.set(buffer_instance);

843

76

844

130

    return buffer_.set(response_body_buffer_);

845

6

  case WasmBufferType::NetworkDownstreamData:

846

6

    return buffer_.set(network_downstream_data_buffer_);

847

3

  case WasmBufferType::NetworkUpstreamData:

848

3

    return buffer_.set(network_upstream_data_buffer_);

849

5

  case WasmBufferType::HttpCallResponseBody:

850

5

    response = rootContext()->http_call_response_;

851

5

    if (response) {

852

3

      auto& body = (*response)->body();

853

3

      return buffer_.set(

854

3

          std::string_view(static_cast<const char*>(body.linearize(body.length())), body.length()));

855

3

856

2

    return nullptr;

857

25

  case WasmBufferType::GrpcReceiveBuffer:

858

25

    return buffer_.set(rootContext()->grpc_receive_buffer_.get());

859

  default:

860

    return nullptr;

861

839

862

839

863

864

27

void Context::onDownstreamConnectionClose(CloseType close_type) {

865

27

  ContextBase::onDownstreamConnectionClose(close_type);

866

27

  downstream_closed_ = true;

867

27

  onCloseTCP();

868

27

869

870

3

void Context::onUpstreamConnectionClose(CloseType close_type) {

871

3

  ContextBase::onUpstreamConnectionClose(close_type);

872

3

  upstream_closed_ = true;

873

3

  if (downstream_closed_) {

874

    onCloseTCP();

875

876

3

877

878

// Async call via HTTP

879

WasmResult Context::httpCall(std::string_view cluster, const Pairs& request_headers,

880

                             std::string_view request_body, const Pairs& request_trailers,

881

58

                             int timeout_milliseconds, uint32_t* token_ptr) {

882

58

  if (timeout_milliseconds < 0) {

883

11

    return WasmResult::BadArgument;

884

11

885

47

  auto cluster_string = std::string(cluster);

886

47

  const auto thread_local_cluster = clusterManager().getThreadLocalCluster(cluster_string);

887

47

  if (thread_local_cluster == nullptr) {

888

11

    return WasmResult::BadArgument;

889

11

890

891

36

  Http::RequestMessagePtr message(

892

36

      new Http::RequestMessageImpl(buildRequestHeaderMapFromPairs(request_headers)));

893

894

  // Check that we were provided certain headers.

895

36

  if (message->headers().Path() == nullptr || message->headers().Method() == nullptr ||

896

36

      message->headers().Host() == nullptr) {

897

11

    return WasmResult::BadArgument;

898

11

899

900

25

  if (!request_body.empty()) {

901

24

    message->body().add(toAbslStringView(request_body));

902

24

    message->headers().setContentLength(request_body.size());

903

24

904

905

25

  if (!request_trailers.empty()) {

906

15

    message->trailers(buildRequestTrailerMapFromPairs(request_trailers));

907

15

908

909

25

  absl::optional<std::chrono::milliseconds> timeout;

910

25

  if (timeout_milliseconds > 0) {

911

25

    timeout = std::chrono::milliseconds(timeout_milliseconds);

912

25

913

914

25

  uint32_t token = envoyWasm()->nextHttpCallId();

915

25

  auto& handler = http_request_[token];

916

25

  handler.context_ = this;

917

25

  handler.token_ = token;

918

919

  // set default hash policy to be based on :authority to enable consistent hash

920

25

  Http::AsyncClient::RequestOptions options;

921

25

  options.setTimeout(timeout);

922

25

  Protobuf::RepeatedPtrField<HashPolicy> hash_policy;

923

25

  hash_policy.Add()->mutable_header()->set_header_name(Http::Headers::get().Host.get());

924

25

  options.setHashPolicy(hash_policy);

925

25

  options.setSendXff(false);

926

25

  auto http_request =

927

25

      thread_local_cluster->httpAsyncClient().send(std::move(message), handler, options);

928

25

  if (!http_request) {

929

6

    http_request_.erase(token);

930

6

    return WasmResult::InternalFailure;

931

6

932

19

  handler.request_ = http_request;

933

19

  *token_ptr = token;

934

19

  return WasmResult::Ok;

935

25

936

937

WasmResult Context::grpcCall(std::string_view grpc_service, std::string_view service_name,

938

                             std::string_view method_name, const Pairs& initial_metadata,

939

                             std::string_view request, std::chrono::milliseconds timeout,

940

60

                             uint32_t* token_ptr) {

941

60

  GrpcService service_proto;

942

60

  if (!service_proto.ParseFromString(grpc_service)) {

943

48

    auto cluster_name = std::string(grpc_service.substr(0, grpc_service.size()));

944

48

    const auto thread_local_cluster = clusterManager().getThreadLocalCluster(cluster_name);

945

48

    if (thread_local_cluster == nullptr) {

946

      // TODO(shikugawa): The reason to keep return status as `BadArgument` is not to force

947

      // callers to change their own codebase with ABI 0.1.x. We should treat this failure as

948

      // `BadArgument` after ABI 0.2.x will have released.

949

30

      return WasmResult::ParseFailure;

950

30

951

18

    service_proto.mutable_envoy_grpc()->set_cluster_name(cluster_name);

952

18

953

30

  uint32_t token = envoyWasm()->nextGrpcCallId();

954

30

  auto& handler = grpc_call_request_[token];

955

30

  handler.context_ = this;

956

30

  handler.token_ = token;

957

30

  auto client_or_error = clusterManager().grpcAsyncClientManager().getOrCreateRawAsyncClient(

958

30

      service_proto, *envoyWasm()->scope_, true /* skip_cluster_check */);

959

30

  if (!client_or_error.status().ok()) {

960

    return WasmResult::BadArgument;

961

962

30

  auto grpc_client = client_or_error.value();

963

30

  grpc_initial_metadata_ = buildRequestHeaderMapFromPairs(initial_metadata);

964

965

  // set default hash policy to be based on :authority to enable consistent hash

966

30

  Http::AsyncClient::RequestOptions options;

967

30

  options.setTimeout(timeout);

968

30

  Protobuf::RepeatedPtrField<HashPolicy> hash_policy;

969

30

  hash_policy.Add()->mutable_header()->set_header_name(Http::Headers::get().Host.get());

970

30

  options.setHashPolicy(hash_policy);

971

30

  options.setSendXff(false);

972

973

30

  auto grpc_request =

974

30

      grpc_client->sendRaw(toAbslStringView(service_name), toAbslStringView(method_name),

975

30

                           std::make_unique<::Envoy::Buffer::OwnedImpl>(toAbslStringView(request)),

976

30

                           handler, Tracing::NullSpan::instance(), options);

977

30

  if (!grpc_request) {

978

5

    grpc_call_request_.erase(token);

979

5

    return WasmResult::InternalFailure;

980

5

981

25

  handler.client_ = std::move(grpc_client);

982

25

  handler.request_ = grpc_request;

983

25

  *token_ptr = token;

984

25

  return WasmResult::Ok;

985

30

986

987

WasmResult Context::grpcStream(std::string_view grpc_service, std::string_view service_name,

988

                               std::string_view method_name, const Pairs& initial_metadata,

989

75

                               uint32_t* token_ptr) {

990

75

  GrpcService service_proto;

991

75

  if (!service_proto.ParseFromString(grpc_service)) {

992

55

    auto cluster_name = std::string(grpc_service.substr(0, grpc_service.size()));

993

55

    const auto thread_local_cluster = clusterManager().getThreadLocalCluster(cluster_name);

994

55

    if (thread_local_cluster == nullptr) {

995

      // TODO(shikugawa): The reason to keep return status as `BadArgument` is not to force

996

      // callers to change their own codebase with ABI 0.1.x. We should treat this failure as

997

      // `BadArgument` after ABI 0.2.x will have released.

998

25

      return WasmResult::ParseFailure;

999

25

1000

30

    service_proto.mutable_envoy_grpc()->set_cluster_name(cluster_name);

1001

30

1002

50

  uint32_t token = envoyWasm()->nextGrpcStreamId();

1003

50

  auto& handler = grpc_stream_[token];

1004

50

  handler.context_ = this;

1005

50

  handler.token_ = token;

1006

50

  auto client_or_error = clusterManager().grpcAsyncClientManager().getOrCreateRawAsyncClient(

1007

50

      service_proto, *envoyWasm()->scope_, true /* skip_cluster_check */);

1008

50

  if (!client_or_error.status().ok()) {

1009

    return WasmResult::BadArgument;

1010

1011

50

  auto grpc_client = client_or_error.value();

1012

50

  grpc_initial_metadata_ = buildRequestHeaderMapFromPairs(initial_metadata);

1013

1014

  // set default hash policy to be based on :authority to enable consistent hash

1015

50

  Http::AsyncClient::StreamOptions options;

1016

50

  Protobuf::RepeatedPtrField<HashPolicy> hash_policy;

1017

50

  hash_policy.Add()->mutable_header()->set_header_name(Http::Headers::get().Host.get());

1018

50

  options.setHashPolicy(hash_policy);

1019

50

  options.setSendXff(false);

1020

1021

50

  auto grpc_stream = grpc_client->startRaw(toAbslStringView(service_name),

1022

50

                                           toAbslStringView(method_name), handler, options);

1023

50

  if (!grpc_stream) {

1024

25

    grpc_stream_.erase(token);

1025

25

    return WasmResult::InternalFailure;

1026

25

1027

25

  handler.client_ = std::move(grpc_client);

1028

25

  handler.stream_ = grpc_stream;

1029

25

  *token_ptr = token;

1030

25

  return WasmResult::Ok;

1031

50

1032

1033

// NB: this is currently called inline, so the token is known to be that of the currently

1034

// executing grpcCall or grpcStream.

1035

void Context::onGrpcCreateInitialMetadata(uint32_t /* token */,

1036

30

                                          Http::RequestHeaderMap& initial_metadata) {

1037

30

  if (grpc_initial_metadata_) {

1038

30

    Http::HeaderMapImpl::copyFrom(initial_metadata, *grpc_initial_metadata_);

1039

30

    grpc_initial_metadata_.reset();

1040

30

1041

30

1042

1043

// StreamInfo

1044

745

const StreamInfo::StreamInfo* Context::getConstRequestStreamInfo() const {

1045

745

  if (encoder_callbacks_) {

1046

88

    return &encoder_callbacks_->streamInfo();

1047

657

  } else if (decoder_callbacks_) {

1048

1

    return &decoder_callbacks_->streamInfo();

1049

656

  } else if (access_log_stream_info_) {

1050

1

    return access_log_stream_info_;

1051

655

  } else if (network_read_filter_callbacks_) {

1052

1

    return &network_read_filter_callbacks_->connection().streamInfo();

1053

654

  } else if (network_write_filter_callbacks_) {

1054

1

    return &network_write_filter_callbacks_->connection().streamInfo();

1055

1

1056

653

  return nullptr;

1057

745

1058

1059

25

StreamInfo::StreamInfo* Context::getRequestStreamInfo() const {

1060

25

  if (encoder_callbacks_) {

1061

16

    return &encoder_callbacks_->streamInfo();

1062

22

  } else if (decoder_callbacks_) {

1063

1

    return &decoder_callbacks_->streamInfo();

1064

8

  } else if (network_read_filter_callbacks_) {

1065

4

    return &network_read_filter_callbacks_->connection().streamInfo();

1066

7

  } else if (network_write_filter_callbacks_) {

1067

1

    return &network_write_filter_callbacks_->connection().streamInfo();

1068

1

1069

3

  return nullptr;

1070

25

1071

1072

11

const Network::Connection* Context::getConnection() const {

1073

11

  if (encoder_callbacks_) {

1074

3

    return encoder_callbacks_->connection().ptr();

1075

8

  } else if (decoder_callbacks_) {

1076

1

    return decoder_callbacks_->connection().ptr();

1077

7

  } else if (network_read_filter_callbacks_) {

1078

1

    return &network_read_filter_callbacks_->connection();

1079

6

  } else if (network_write_filter_callbacks_) {

1080

1

    return &network_write_filter_callbacks_->connection();

1081

1

1082

5

  return nullptr;

1083

11

1084

1085

17

WasmResult Context::setProperty(std::string_view path, std::string_view value) {

1086

17

  auto* stream_info = getRequestStreamInfo();

1087

17

  if (!stream_info) {

1088

2

    return WasmResult::NotFound;

1089

2

1090

15

  std::string key;

1091

15

  absl::StrAppend(&key, CelStateKeyPrefix, toAbslStringView(path));

1092

15

  CelState* state = stream_info->filterState()->getDataMutable<CelState>(key);

1093

15

  if (state == nullptr) {

1094

13

    const auto& it = rootContext()->state_prototypes_.find(toAbslStringView(path));

1095

13

    const CelStatePrototype& prototype =

1096

13

        it == rootContext()->state_prototypes_.end()

1097

13

            ? Filters::Common::Expr::DefaultCelStatePrototype::get()

1098

13

            : *it->second.get(); // NOLINT

1099

13

    auto state_ptr = std::make_unique<CelState>(prototype);

1100

13

    state = state_ptr.get();

1101

13

    stream_info->filterState()->setData(key, std::move(state_ptr),

1102

13

                                        StreamInfo::FilterState::StateType::Mutable,

1103

13

                                        prototype.life_span_);

1104

13

1105

15

  if (!state->setValue(toAbslStringView(value))) {

1106

2

    return WasmResult::BadArgument;

1107

2

1108

13

  return WasmResult::Ok;

1109

15

1110

1111

WasmResult Context::setEnvoyFilterState(std::string_view path, std::string_view value,

1112

3

                                        StreamInfo::FilterState::LifeSpan life_span) {

1113

3

  auto* factory =

1114

3

      Registry::FactoryRegistry<StreamInfo::FilterState::ObjectFactory>::getFactory(path);

1115

3

  if (!factory) {

1116

1

    return WasmResult::NotFound;

1117

1

1118

1119

2

  auto object = factory->createFromBytes(value);

1120

2

  if (!object) {

1121

    return WasmResult::BadArgument;

1122

1123

1124

2

  auto* stream_info = getRequestStreamInfo();

1125

2

  if (!stream_info) {

1126

    return WasmResult::NotFound;

1127

1128

1129

2

  stream_info->filterState()->setData(path, std::move(object),

1130

2

                                      StreamInfo::FilterState::StateType::Mutable, life_span);

1131

2

  return WasmResult::Ok;

1132

2

1133

1134

WasmResult

1135

Context::declareProperty(std::string_view path,

1136

10

                         Filters::Common::Expr::CelStatePrototypeConstPtr state_prototype) {

1137

  // Do not delete existing schema since it can be referenced by state objects.

1138

10

  if (state_prototypes_.find(toAbslStringView(path)) == state_prototypes_.end()) {

1139

8

    state_prototypes_[toAbslStringView(path)] = std::move(state_prototype);

1140

8

    return WasmResult::Ok;

1141

8

1142

2

  return WasmResult::BadArgument;

1143

10

1144

1145

192

WasmResult Context::log(uint32_t level, std::string_view message) {

1146

192

  switch (static_cast<spdlog::level::level_enum>(level)) {

1147

7

  case spdlog::level::trace:

1148

7

    ENVOY_LOG(trace, "wasm log{}: {}", log_prefix(), message);

1149

7

    return WasmResult::Ok;

1150

26

  case spdlog::level::debug:

1151

26

    ENVOY_LOG(debug, "wasm log{}: {}", log_prefix(), message);

1152

26

    return WasmResult::Ok;

1153

52

  case spdlog::level::info:

1154

52

    ENVOY_LOG(info, "wasm log{}: {}", log_prefix(), message);

1155

52

    return WasmResult::Ok;

1156

37

  case spdlog::level::warn:

1157

37

    ENVOY_LOG(warn, "wasm log{}: {}", log_prefix(), message);

1158

37

    return WasmResult::Ok;

1159

70

  case spdlog::level::err:

1160

70

    ENVOY_LOG(error, "wasm log{}: {}", log_prefix(), message);

1161

70

    return WasmResult::Ok;

1162

  case spdlog::level::critical:

1163

    ENVOY_LOG(critical, "wasm log{}: {}", log_prefix(), message);

1164

    return WasmResult::Ok;

1165

  case spdlog::level::off:

1166

    PANIC("not implemented");

1167

  case spdlog::level::n_levels:

1168

    PANIC("not implemented");

1169

192

1170

  PANIC_DUE_TO_CORRUPT_ENUM;

1171

1172

1173

2

uint32_t Context::getLogLevel() {

1174

  // Like the "log" call above, assume that spdlog level as an int

1175

  // matches the enum in the SDK

1176

2

  return static_cast<uint32_t>(ENVOY_LOGGER().level());

1177

2

1178

1179

//

1180

// Calls into the Wasm code.

1181

//

1182

bool Context::validateConfiguration(std::string_view configuration,

1183

3

                                    const std::shared_ptr<PluginBase>& plugin_base) {

1184

3

  auto plugin = std::static_pointer_cast<Plugin>(plugin_base);

1185

3

  if (!envoyWasm()->validate_configuration_) {

1186

1

    return true;

1187

1

1188

2

  temp_plugin_ = plugin_base;

1189

2

  auto result =

1190

2

      envoyWasm()

1191

2

          ->validate_configuration_(this, id_, static_cast<uint32_t>(configuration.size()))

1192

2

          .u64_ != 0;

1193

2

  temp_plugin_.reset();

1194

2

  return result;

1195

3

1196

1197

2

std::string_view Context::getConfiguration() {

1198

2

  if (temp_plugin_) {

1199

    return temp_plugin_->plugin_configuration_;

1200

2

  } else {

1201

2

    return envoyWasm()->vm_configuration();

1202

2

1203

2

};

1204

1205

22

std::pair<uint32_t, std::string_view> Context::getStatus() {

1206

22

  return std::make_pair(status_code_, toStdStringView(status_message_));

1207

22

1208

1209

25

void Context::onGrpcReceiveInitialMetadataWrapper(uint32_t token, Http::HeaderMapPtr&& metadata) {

1210

25

  grpc_receive_initial_metadata_ = std::move(metadata);

1211

25

  onGrpcReceiveInitialMetadata(token, headerSize(grpc_receive_initial_metadata_));

1212

25

  grpc_receive_initial_metadata_ = nullptr;

1213

25

1214

1215

10

void Context::onGrpcReceiveTrailingMetadataWrapper(uint32_t token, Http::HeaderMapPtr&& metadata) {

1216

10

  grpc_receive_trailing_metadata_ = std::move(metadata);

1217

10

  onGrpcReceiveTrailingMetadata(token, headerSize(grpc_receive_trailing_metadata_));

1218

10

  grpc_receive_trailing_metadata_ = nullptr;

1219

10

1220

1221

WasmResult Context::defineMetric(uint32_t metric_type, std::string_view name,

1222

34

                                 uint32_t* metric_id_ptr) {

1223

34

  if (metric_type > static_cast<uint32_t>(MetricType::Max)) {

1224

2

    return WasmResult::BadArgument;

1225

2

1226

32

  auto type = static_cast<MetricType>(metric_type);

1227

  // TODO: Consider rethinking the scoping policy as it does not help in this case.

1228

32

  Stats::StatNameManagedStorage storage(toAbslStringView(name), envoyWasm()->scope_->symbolTable());

1229

32

  Stats::StatName stat_name = storage.statName();

1230

  // We prefix the given name with custom_stat_name_ so that these user-defined

1231

  // custom metrics can be distinguished from native Envoy metrics.

1232

32

  if (type == MetricType::Counter) {

1233

12

    auto id = envoyWasm()->nextCounterMetricId();

1234

12

    Stats::Counter* c = &Stats::Utility::counterFromElements(

1235

12

        *envoyWasm()->scope_, {envoyWasm()->custom_stat_namespace_, stat_name});

1236

12

    envoyWasm()->counters_.emplace(id, c);

1237

12

    *metric_id_ptr = id;

1238

12

    return WasmResult::Ok;

1239

12

1240

20

  if (type == MetricType::Gauge) {

1241

10

    auto id = envoyWasm()->nextGaugeMetricId();

1242

10

    Stats::Gauge* g = &Stats::Utility::gaugeFromStatNames(

1243

10

        *envoyWasm()->scope_, {envoyWasm()->custom_stat_namespace_, stat_name},

1244

10

        Stats::Gauge::ImportMode::Accumulate);

1245

10

    envoyWasm()->gauges_.emplace(id, g);

1246

10

    *metric_id_ptr = id;

1247

10

    return WasmResult::Ok;

1248

10

1249

  // (type == MetricType::Histogram) {

1250

10

  auto id = envoyWasm()->nextHistogramMetricId();

1251

10

  Stats::Histogram* h = &Stats::Utility::histogramFromStatNames(

1252

10

      *envoyWasm()->scope_, {envoyWasm()->custom_stat_namespace_, stat_name},

1253

10

      Stats::Histogram::Unit::Unspecified);

1254

10

  envoyWasm()->histograms_.emplace(id, h);

1255

10

  *metric_id_ptr = id;

1256

10

  return WasmResult::Ok;

1257

20

1258

1259

34

WasmResult Context::incrementMetric(uint32_t metric_id, int64_t offset) {

1260

34

  auto type = static_cast<MetricType>(metric_id & Wasm::kMetricTypeMask);

1261

34

  if (type == MetricType::Counter) {

1262

22

    auto it = envoyWasm()->counters_.find(metric_id);

1263

22

    if (it != envoyWasm()->counters_.end()) {

1264

20

      if (offset > 0) {

1265

18

        it->second->add(offset);

1266

18

        return WasmResult::Ok;

1267

18

      } else {

1268

2

        return WasmResult::BadArgument;

1269

2

1270

20

1271

2

    return WasmResult::NotFound;

1272

26

  } else if (type == MetricType::Gauge) {

1273

6

    auto it = envoyWasm()->gauges_.find(metric_id);

1274

6

    if (it != envoyWasm()->gauges_.end()) {

1275

4

      if (offset > 0) {

1276

2

        it->second->add(offset);

1277

2

        return WasmResult::Ok;

1278

2

      } else {

1279

2

        it->second->sub(-offset);

1280

2

        return WasmResult::Ok;

1281

2

1282

4

1283

2

    return WasmResult::NotFound;

1284

6

1285

6

  return WasmResult::BadArgument;

1286

34

1287

1288

36

WasmResult Context::recordMetric(uint32_t metric_id, uint64_t value) {

1289

36

  auto type = static_cast<MetricType>(metric_id & Wasm::kMetricTypeMask);

1290

36

  if (type == MetricType::Counter) {

1291

10

    auto it = envoyWasm()->counters_.find(metric_id);

1292

10

    if (it != envoyWasm()->counters_.end()) {

1293

8

      it->second->add(value);

1294

8

      return WasmResult::Ok;

1295

8

1296

26

  } else if (type == MetricType::Gauge) {

1297

12

    auto it = envoyWasm()->gauges_.find(metric_id);

1298

12

    if (it != envoyWasm()->gauges_.end()) {

1299

10

      it->second->set(value);

1300

10

      return WasmResult::Ok;

1301

10

1302

14

  } else if (type == MetricType::Histogram) {

1303

12

    auto it = envoyWasm()->histograms_.find(metric_id);

1304

12

    if (it != envoyWasm()->histograms_.end()) {

1305

10

      it->second->recordValue(value);

1306

10

      return WasmResult::Ok;

1307

10

1308

12

1309

8

  return WasmResult::NotFound;

1310

36

1311

1312

48

WasmResult Context::getMetric(uint32_t metric_id, uint64_t* result_uint64_ptr) {

1313

48

  auto type = static_cast<MetricType>(metric_id & Wasm::kMetricTypeMask);

1314

48

  if (type == MetricType::Counter) {

1315

28

    auto it = envoyWasm()->counters_.find(metric_id);

1316

28

    if (it != envoyWasm()->counters_.end()) {

1317

26

      *result_uint64_ptr = it->second->value();

1318

26

      return WasmResult::Ok;

1319

26

1320

2

    return WasmResult::NotFound;

1321

30

  } else if (type == MetricType::Gauge) {

1322

12

    auto it = envoyWasm()->gauges_.find(metric_id);

1323

12

    if (it != envoyWasm()->gauges_.end()) {

1324

10

      *result_uint64_ptr = it->second->value();

1325

10

      return WasmResult::Ok;

1326

10

1327

2

    return WasmResult::NotFound;

1328

12

1329

8

  return WasmResult::BadArgument;

1330

48

1331

1332

2139

Context::~Context() {

1333

  // Cancel any outstanding requests.

1334

2139

  for (auto& p : http_request_) {

1335

3

    if (p.second.request_ != nullptr) {

1336

3

      p.second.request_->cancel();

1337

3

1338

3

1339

2139

  for (auto& p : grpc_call_request_) {

1340

1

    if (p.second.request_ != nullptr) {

1341

1

      p.second.request_->cancel();

1342

1

1343

1

1344

2139

  for (auto& p : grpc_stream_) {

1345

5

    if (p.second.stream_ != nullptr) {

1346

5

      p.second.stream_->resetStream();

1347

5

1348

5

1349

2139

1350

1351

53

Network::FilterStatus convertNetworkFilterStatus(proxy_wasm::FilterStatus status) {

1352

53

  switch (status) {

1353

  default:

1354

37

  case proxy_wasm::FilterStatus::Continue:

1355

37

    return Network::FilterStatus::Continue;

1356

16

  case proxy_wasm::FilterStatus::StopIteration:

1357

16

    return Network::FilterStatus::StopIteration;

1358

53

1359

53

};

1360

1361

226

Http::FilterHeadersStatus convertFilterHeadersStatus(proxy_wasm::FilterHeadersStatus status) {

1362

226

  switch (status) {

1363

  default:

1364

138

  case proxy_wasm::FilterHeadersStatus::Continue:

1365

138

    return Http::FilterHeadersStatus::Continue;

1366

2

  case proxy_wasm::FilterHeadersStatus::StopIteration:

1367

2

    return Http::FilterHeadersStatus::StopIteration;

1368

2

  case proxy_wasm::FilterHeadersStatus::StopAllIterationAndBuffer:

1369

2

    return Http::FilterHeadersStatus::StopAllIterationAndBuffer;

1370

84

  case proxy_wasm::FilterHeadersStatus::StopAllIterationAndWatermark:

1371

84

    return Http::FilterHeadersStatus::StopAllIterationAndWatermark;

1372

226

1373

226

};

1374

1375

6

Http::FilterTrailersStatus convertFilterTrailersStatus(proxy_wasm::FilterTrailersStatus status) {

1376

6

  switch (status) {

1377

  default:

1378

3

  case proxy_wasm::FilterTrailersStatus::Continue:

1379

3

    return Http::FilterTrailersStatus::Continue;

1380

3

  case proxy_wasm::FilterTrailersStatus::StopIteration:

1381

3

    return Http::FilterTrailersStatus::StopIteration;

1382

6

1383

6

};

1384

1385

6

Http::FilterMetadataStatus convertFilterMetadataStatus(proxy_wasm::FilterMetadataStatus status) {

1386

6

  switch (status) {

1387

  default:

1388

6

  case proxy_wasm::FilterMetadataStatus::Continue:

1389

6

    return Http::FilterMetadataStatus::Continue;

1390

6

1391

6

};

1392

1393

195

Http::FilterDataStatus convertFilterDataStatus(proxy_wasm::FilterDataStatus status) {

1394

195

  switch (status) {

1395

  default:

1396

109

  case proxy_wasm::FilterDataStatus::Continue:

1397

109

    return Http::FilterDataStatus::Continue;

1398

60

  case proxy_wasm::FilterDataStatus::StopIterationAndBuffer:

1399

60

    return Http::FilterDataStatus::StopIterationAndBuffer;

1400

4

  case proxy_wasm::FilterDataStatus::StopIterationAndWatermark:

1401

4

    return Http::FilterDataStatus::StopIterationAndWatermark;

1402

22

  case proxy_wasm::FilterDataStatus::StopIterationNoBuffer:

1403

22

    return Http::FilterDataStatus::StopIterationNoBuffer;

1404

195

1405

195

};

1406

1407

30

Network::FilterStatus Context::onNewConnection() {

1408

30

  onCreate();

1409

30

  return convertNetworkFilterStatus(onNetworkNewConnection());

1410

30

};

1411

1412

15

Network::FilterStatus Context::onData(::Envoy::Buffer::Instance& data, bool end_stream) {

1413

15

  if (!in_vm_context_created_) {

1414

3

    return Network::FilterStatus::Continue;

1415

3

1416

12

  network_downstream_data_buffer_ = &data;

1417

12

  end_of_stream_ = end_stream;

1418

12

  auto result = convertNetworkFilterStatus(onDownstreamData(data.length(), end_stream));

1419

12

  if (result == Network::FilterStatus::Continue) {

1420

7

    network_downstream_data_buffer_ = nullptr;

1421

7

1422

12

  return result;

1423

15

1424

1425

14

Network::FilterStatus Context::onWrite(::Envoy::Buffer::Instance& data, bool end_stream) {

1426

14

  if (!in_vm_context_created_) {

1427

3

    return Network::FilterStatus::Continue;

1428

3

1429

11

  network_upstream_data_buffer_ = &data;

1430

11

  end_of_stream_ = end_stream;

1431

11

  auto result = convertNetworkFilterStatus(onUpstreamData(data.length(), end_stream));

1432

11

  if (result == Network::FilterStatus::Continue) {

1433

6

    network_upstream_data_buffer_ = nullptr;

1434

6

1435

11

  if (end_stream) {

1436

    // This is called when seeing end_stream=true and not on an upstream connection event,

1437

    // because registering for latter requires replicating the whole TCP proxy extension.

1438

3

    onUpstreamConnectionClose(CloseType::Unknown);

1439

3

1440

11

  return result;

1441

14

1442

1443

33

void Context::onEvent(Network::ConnectionEvent event) {

1444

33

  if (!in_vm_context_created_) {

1445

3

    return;

1446

3

1447

30

  switch (event) {

1448

24

  case Network::ConnectionEvent::LocalClose:

1449

24

    onDownstreamConnectionClose(CloseType::Local);

1450

24

    break;

1451

3

  case Network::ConnectionEvent::RemoteClose:

1452

3

    onDownstreamConnectionClose(CloseType::Remote);

1453

3

    break;

1454

3

  default:

1455

3

    break;

1456

30

1457

30

1458

1459

41

void Context::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {

1460

41

  network_read_filter_callbacks_ = &callbacks;

1461

41

  network_read_filter_callbacks_->connection().addConnectionCallbacks(*this);

1462

41

1463

1464

40

void Context::initializeWriteFilterCallbacks(Network::WriteFilterCallbacks& callbacks) {

1465

40

  network_write_filter_callbacks_ = &callbacks;

1466

40

1467

1468

void Context::log(const Formatter::Context& log_context,

1469

36

                  const StreamInfo::StreamInfo& stream_info) {

1470

  // `log` may be called multiple times due to mid-request logging -- we only want to run on the

1471

  // last call.

1472

36

  if (!stream_info.requestComplete().has_value()) {

1473

3

    return;

1474

3

1475

33

  if (!in_vm_context_created_) {

1476

    // If the request is invalid then onRequestHeaders() will not be called and neither will

1477

    // onCreate() in cases like sendLocalReply who short-circuits envoy

1478

    // lifecycle. This is because Envoy does not have a well defined lifetime for the combined

1479

    // HTTP

1480

    // + AccessLog filter. Thus, to log these scenarios, we call onCreate() in log function below.

1481

5

    onCreate();

1482

5

1483

1484

33

  access_log_phase_ = true;

1485

33

  access_log_request_headers_ = log_context.requestHeaders().ptr();

1486

  // ? request_trailers  ?

1487

33

  access_log_response_headers_ = log_context.responseHeaders().ptr();

1488

33

  access_log_response_trailers_ = log_context.responseTrailers().ptr();

1489

33

  access_log_stream_info_ = &stream_info;

1490

1491

33

  onLog();

1492

1493

33

  access_log_phase_ = false;

1494

33

  access_log_request_headers_ = nullptr;

1495

  // ? request_trailers  ?

1496

33

  access_log_response_headers_ = nullptr;

1497

33

  access_log_response_trailers_ = nullptr;

1498

33

  access_log_stream_info_ = nullptr;

1499

33

1500

1501

113

void Context::onDestroy() {

1502

113

  if (destroyed_ || !in_vm_context_created_) {

1503

6

    return;

1504

6

1505

107

  destroyed_ = true;

1506

107

  onDone();

1507

107

  onDelete();

1508

107

1509

1510

22

WasmResult Context::continueStream(WasmStreamType stream_type) {

1511

22

  switch (stream_type) {

1512

11

  case WasmStreamType::Request:

1513

11

    if (decoder_callbacks_) {

1514

      // We are in a reentrant call, so defer.

1515

11

      envoyWasm()->addAfterVmCallAction([this] { decoder_callbacks_->continueDecoding(); });

1516

11

1517

11

    break;

1518

3

  case WasmStreamType::Response:

1519

3

    if (encoder_callbacks_) {

1520

      // We are in a reentrant call, so defer.

1521

3

      envoyWasm()->addAfterVmCallAction([this] { encoder_callbacks_->continueEncoding(); });

1522

3

1523

3

    break;

1524

3

  case WasmStreamType::Downstream:

1525

3

    if (network_read_filter_callbacks_) {

1526

      // We are in a reentrant call, so defer.

1527

3

      envoyWasm()->addAfterVmCallAction(

1528

3

          [this] { network_read_filter_callbacks_->continueReading(); });

1529

3

1530

3

    return WasmResult::Ok;

1531

2

  case WasmStreamType::Upstream:

1532

2

    return WasmResult::Unimplemented;

1533

3

  default:

1534

3

    return WasmResult::BadArgument;

1535

22

1536

14

  request_headers_ = nullptr;

1537

14

  request_body_buffer_ = nullptr;

1538

14

  request_trailers_ = nullptr;

1539

14

  request_metadata_ = nullptr;

1540

14

  return WasmResult::Ok;

1541

22

1542

1543

constexpr absl::string_view CloseStreamResponseDetails = "wasm_close_stream";

1544

1545

18

WasmResult Context::closeStream(WasmStreamType stream_type) {

1546

18

  switch (stream_type) {

1547

3

  case WasmStreamType::Request:

1548

3

    if (decoder_callbacks_) {

1549

3

      if (!decoder_callbacks_->streamInfo().responseCodeDetails().has_value()) {

1550

3

        decoder_callbacks_->streamInfo().setResponseCodeDetails(CloseStreamResponseDetails);

1551

3

1552

      // We are in a reentrant call, so defer.

1553

3

      envoyWasm()->addAfterVmCallAction([this] { decoder_callbacks_->resetStream(); });

1554

3

1555

3

    return WasmResult::Ok;

1556

6

  case WasmStreamType::Response:

1557

6

    if (encoder_callbacks_) {

1558

6

      if (!encoder_callbacks_->streamInfo().responseCodeDetails().has_value()) {

1559

6

        encoder_callbacks_->streamInfo().setResponseCodeDetails(CloseStreamResponseDetails);

1560

6

1561

      // We are in a reentrant call, so defer.

1562

6

      envoyWasm()->addAfterVmCallAction([this] { encoder_callbacks_->resetStream(); });

1563

6

1564

6

    return WasmResult::Ok;

1565

3

  case WasmStreamType::Downstream:

1566

3

    if (network_read_filter_callbacks_) {

1567

      // We are in a reentrant call, so defer.

1568

3

      envoyWasm()->addAfterVmCallAction([this] {

1569

3

        network_read_filter_callbacks_->connection().close(

1570

3

            Envoy::Network::ConnectionCloseType::FlushWrite, "wasm_downstream_close");

1571

3

});

1572

3

1573

3

    return WasmResult::Ok;

1574

3

  case WasmStreamType::Upstream:

1575

3

    if (network_write_filter_callbacks_) {

1576

      // We are in a reentrant call, so defer.

1577

3

      envoyWasm()->addAfterVmCallAction([this] {

1578

3

        network_write_filter_callbacks_->connection().close(

1579

3

            Envoy::Network::ConnectionCloseType::FlushWrite, "wasm_upstream_close");

1580

3

});

1581

3

1582

3

    return WasmResult::Ok;

1583

18

1584

3

  return WasmResult::BadArgument;

1585

18

1586

1587

constexpr absl::string_view FailStreamResponseDetails = "wasm_fail_stream";

1588

1589

92

void Context::failStream(WasmStreamType stream_type) {

1590

92

  switch (stream_type) {

1591

36

  case WasmStreamType::Request:

1592

36

    if (decoder_callbacks_ && !failure_local_reply_sent_) {

1593

36

      decoder_callbacks_->sendLocalReply(Envoy::Http::Code::ServiceUnavailable, "", nullptr,

1594

36

                                         Grpc::Status::WellKnownGrpcStatus::Unavailable,

1595

36

                                         FailStreamResponseDetails);

1596

36

      failure_local_reply_sent_ = true;

1597

36

1598

36

    break;

1599

36

  case WasmStreamType::Response:

1600

36

    if (encoder_callbacks_ && !failure_local_reply_sent_) {

1601

      encoder_callbacks_->sendLocalReply(Envoy::Http::Code::ServiceUnavailable, "", nullptr,

1602

                                         Grpc::Status::WellKnownGrpcStatus::Unavailable,

1603

                                         FailStreamResponseDetails);

1604

      failure_local_reply_sent_ = true;

1605

1606

36

    break;

1607

10

  case WasmStreamType::Downstream:

1608

10

    if (network_read_filter_callbacks_) {

1609

10

      network_read_filter_callbacks_->connection().close(

1610

10

          Envoy::Network::ConnectionCloseType::FlushWrite);

1611

10

1612

10

    break;

1613

10

  case WasmStreamType::Upstream:

1614

10

    if (network_write_filter_callbacks_) {

1615

10

      network_write_filter_callbacks_->connection().close(

1616

10

          Envoy::Network::ConnectionCloseType::FlushWrite);

1617

10

1618

10

    break;

1619

92

1620

92

1621

1622

WasmResult Context::sendLocalResponse(uint32_t response_code, std::string_view body_text,

1623

                                      Pairs additional_headers, uint32_t grpc_status,

1624

15

                                      std::string_view details) {

1625

  // "additional_headers" is a collection of string_views. These will no longer

1626

  // be valid when "modify_headers" is finally called below, so we must

1627

  // make copies of all the headers.

1628

15

  std::vector<std::pair<Http::LowerCaseString, std::string>> additional_headers_copy;

1629

15

  for (auto& p : additional_headers) {

1630

2

    const Http::LowerCaseString lower_key{std::string(p.first)};

1631

2

    additional_headers_copy.emplace_back(lower_key, std::string(p.second));

1632

2

1633

1634

15

  auto modify_headers = [additional_headers_copy](Http::HeaderMap& headers) {

1635

13

    for (auto& p : additional_headers_copy) {

1636

2

      headers.addCopy(p.first, p.second);

1637

2

1638

13

};

1639

1640

15

  if (decoder_callbacks_) {

1641

    // This is a bit subtle because proxy_on_delete() does call DeferAfterCallActions(),

1642

    // so in theory it could call this and the Context in the VM would be invalid,

1643

    // but because it only gets called after the connections have drained, the call to

1644

    // sendLocalReply() will fail. Net net, this is safe.

1645

15

    envoyWasm()->addAfterVmCallAction([this, response_code, body_text = std::string(body_text),

1646

15

                                       modify_headers = std::move(modify_headers), grpc_status,

1647

15

                                       details = StringUtil::replaceAllEmptySpace(

1648

15

                                           absl::string_view(details.data(), details.size()))] {

1649

      // C++, Rust and other SDKs use -1 (InvalidCode) as the default value if gRPC code is not set,

1650

      // which should be mapped to nullopt in Envoy to prevent it from sending a grpc-status trailer

1651

      // at all.

1652

15

      absl::optional<Grpc::Status::GrpcStatus> grpc_status_code = absl::nullopt;

1653

15

      if (grpc_status >= Grpc::Status::WellKnownGrpcStatus::Ok &&

1654

15

          grpc_status <= Grpc::Status::WellKnownGrpcStatus::MaximumKnown) {

1655

2

        grpc_status_code = Grpc::Status::WellKnownGrpcStatus(grpc_status);

1656

2

1657

15

      decoder_callbacks_->sendLocalReply(static_cast<Envoy::Http::Code>(response_code), body_text,

1658

15

                                         modify_headers, grpc_status_code, details);

1659

15

});

1660

15

1661

15

  return WasmResult::Ok;

1662

15

1663

1664

188

Http::FilterHeadersStatus Context::decodeHeaders(Http::RequestHeaderMap& headers, bool end_stream) {

1665

188

  onCreate();

1666

188

  request_headers_ = &headers;

1667

188

  end_of_stream_ = end_stream;

1668

188

  auto result = convertFilterHeadersStatus(onRequestHeaders(headerSize(&headers), end_stream));

1669

188

  if (result == Http::FilterHeadersStatus::Continue) {

1670

103

    request_headers_ = nullptr;

1671

103

1672

188

  return result;

1673

188

1674

1675

133

Http::FilterDataStatus Context::decodeData(::Envoy::Buffer::Instance& data, bool end_stream) {

1676

133

  if (!in_vm_context_created_) {

1677

3

    return Http::FilterDataStatus::Continue;

1678

3

1679

130

  if (buffering_request_body_) {

1680

38

    decoder_callbacks_->addDecodedData(data, false);

1681

38

    if (destroyed_) {

1682

      // The data adding have triggered a local reply (413) and we needn't to continue to

1683

      // call the VM.

1684

      // Note this is not perfect way. If the local reply processing is stopped by other

1685

      // filters, this filter will still try to call the VM. But at least we can ensure

1686

      // the VM has valid context.

1687

3

      return Http::FilterDataStatus::StopIterationAndBuffer;

1688

3

1689

38

1690

127

  request_body_buffer_ = &data;

1691

127

  end_of_stream_ = end_stream;

1692

127

  const auto buffer = getBuffer(WasmBufferType::HttpRequestBody);

1693

127

  const auto buffer_size = (buffer == nullptr) ? 0 : buffer->size();

1694

127

  auto result = convertFilterDataStatus(onRequestBody(buffer_size, end_stream));

1695

127

  buffering_request_body_ = false;

1696

127

  switch (result) {

1697

65

  case Http::FilterDataStatus::Continue:

1698

65

    request_body_buffer_ = nullptr;

1699

65

    break;

1700

40

  case Http::FilterDataStatus::StopIterationAndBuffer:

1701

40

    buffering_request_body_ = true;

1702

40

    break;

1703

2

  case Http::FilterDataStatus::StopIterationAndWatermark:

1704

22

  case Http::FilterDataStatus::StopIterationNoBuffer:

1705

22

    break;

1706

127

1707

127

  return result;

1708

127

1709

1710

6

Http::FilterTrailersStatus Context::decodeTrailers(Http::RequestTrailerMap& trailers) {

1711

6

  if (!in_vm_context_created_) {

1712

3

    return Http::FilterTrailersStatus::Continue;

1713

3

1714

3

  request_trailers_ = &trailers;

1715

3

  auto result = convertFilterTrailersStatus(onRequestTrailers(headerSize(&trailers)));

1716

3

  if (result == Http::FilterTrailersStatus::Continue) {

1717

3

    request_trailers_ = nullptr;

1718

3

1719

3

  return result;

1720

6

1721

1722

6

Http::FilterMetadataStatus Context::decodeMetadata(Http::MetadataMap& request_metadata) {

1723

6

  if (!in_vm_context_created_) {

1724

3

    return Http::FilterMetadataStatus::Continue;

1725

3

1726

3

  request_metadata_ = &request_metadata;

1727

3

  auto result = convertFilterMetadataStatus(onRequestMetadata(headerSize(&request_metadata)));

1728

3

  if (result == Http::FilterMetadataStatus::Continue) {

1729

3

    request_metadata_ = nullptr;

1730

3

1731

3

  return result;

1732

6

1733

1734

266

void Context::setDecoderFilterCallbacks(Envoy::Http::StreamDecoderFilterCallbacks& callbacks) {

1735

266

  decoder_callbacks_ = &callbacks;

1736

266

1737

1738

3

Http::Filter1xxHeadersStatus Context::encode1xxHeaders(Http::ResponseHeaderMap&) {

1739

3

  return Http::Filter1xxHeadersStatus::Continue;

1740

3

1741

1742

Http::FilterHeadersStatus Context::encodeHeaders(Http::ResponseHeaderMap& headers,

1743

43

                                                 bool end_stream) {

1744

  // If the vm context is not created or the stream has failed and the local reply has been sent,

1745

  // we should not continue to call the VM.

1746

43

  if (!in_vm_context_created_ || failure_local_reply_sent_) {

1747

5

    return Http::FilterHeadersStatus::Continue;

1748

5

1749

38

  response_headers_ = &headers;

1750

38

  end_of_stream_ = end_stream;

1751

38

  auto result = convertFilterHeadersStatus(onResponseHeaders(headerSize(&headers), end_stream));

1752

38

  if (result == Http::FilterHeadersStatus::Continue) {

1753

35

    response_headers_ = nullptr;

1754

35

1755

38

  return result;

1756

43

1757

1758

72

Http::FilterDataStatus Context::encodeData(::Envoy::Buffer::Instance& data, bool end_stream) {

1759

  // If the vm context is not created or the stream has failed and the local reply has been sent,

1760

  // we should not continue to call the VM.

1761

72

  if (!in_vm_context_created_ || failure_local_reply_sent_) {

1762

4

    return Http::FilterDataStatus::Continue;

1763

4

1764

68

  if (buffering_response_body_) {

1765

20

    encoder_callbacks_->addEncodedData(data, false);

1766

20

    if (destroyed_) {

1767

      // The data adding have triggered a local reply (413) and we needn't to continue to

1768

      // call the VM.

1769

      // Note this is not perfect way. If the local reply processing is stopped by other

1770

      // filters, this filter will still try to call the VM. But at least we can ensure

1771

      // the VM has valid context.

1772

      return Http::FilterDataStatus::StopIterationAndBuffer;

1773

1774

20

1775

68

  response_body_buffer_ = &data;

1776

68

  end_of_stream_ = end_stream;

1777

68

  const auto buffer = getBuffer(WasmBufferType::HttpResponseBody);

1778

68

  const auto buffer_size = (buffer == nullptr) ? 0 : buffer->size();

1779

68

  auto result = convertFilterDataStatus(onResponseBody(buffer_size, end_stream));

1780

68

  buffering_response_body_ = false;

1781

68

  switch (result) {

1782

44

  case Http::FilterDataStatus::Continue:

1783

44

    response_body_buffer_ = nullptr;

1784

44

    break;

1785

20

  case Http::FilterDataStatus::StopIterationAndBuffer:

1786

20

    buffering_response_body_ = true;

1787

20

    break;

1788

2

  case Http::FilterDataStatus::StopIterationAndWatermark:

1789

4

  case Http::FilterDataStatus::StopIterationNoBuffer:

1790

4

    break;

1791

68

1792

68

  return result;

1793

68

1794

1795

6

Http::FilterTrailersStatus Context::encodeTrailers(Http::ResponseTrailerMap& trailers) {

1796

  // If the vm context is not created or the stream has failed and the local reply has been sent,

1797

  // we should not continue to call the VM.

1798

6

  if (!in_vm_context_created_ || failure_local_reply_sent_) {

1799

3

    return Http::FilterTrailersStatus::Continue;

1800

3

1801

3

  response_trailers_ = &trailers;

1802

3

  auto result = convertFilterTrailersStatus(onResponseTrailers(headerSize(&trailers)));

1803

3

  if (result == Http::FilterTrailersStatus::Continue) {

1804

    response_trailers_ = nullptr;

1805

1806

3

  return result;

1807

6

1808

1809

6

Http::FilterMetadataStatus Context::encodeMetadata(Http::MetadataMap& response_metadata) {

1810

  // If the vm context is not created or the stream has failed and the local reply has been sent,

1811

  // we should not continue to call the VM.

1812

6

  if (!in_vm_context_created_ || failure_local_reply_sent_) {

1813

3

    return Http::FilterMetadataStatus::Continue;

1814

3

1815

3

  response_metadata_ = &response_metadata;

1816

3

  auto result = convertFilterMetadataStatus(onResponseMetadata(headerSize(&response_metadata)));

1817

3

  if (result == Http::FilterMetadataStatus::Continue) {

1818

3

    response_metadata_ = nullptr;

1819

3

1820

3

  return result;

1821

6

1822

1823

//  Http::FilterMetadataStatus::Continue;

1824

1825

255

void Context::setEncoderFilterCallbacks(Envoy::Http::StreamEncoderFilterCallbacks& callbacks) {

1826

255

  encoder_callbacks_ = &callbacks;

1827

255

1828

1829

31

void Context::onHttpCallSuccess(uint32_t token, Envoy::Http::ResponseMessagePtr&& response) {

1830

  // TODO: convert this into a function in proxy-wasm-cpp-host and use here.

1831

31

  if (proxy_wasm::current_context_ != nullptr) {

1832

    // We are in a reentrant call, so defer.

1833

15

    envoyWasm()->addAfterVmCallAction([this, token, response = response.release()] {

1834

15

      onHttpCallSuccess(token, std::unique_ptr<Envoy::Http::ResponseMessage>(response));

1835

15

});

1836

15

    return;

1837

15

1838

16

  auto handler = http_request_.find(token);

1839

16

  if (handler == http_request_.end()) {

1840

3

    return;

1841

3

1842

13

  http_call_response_ = &response;

1843

13

  uint32_t body_size = response->body().length();

1844

  // Deferred "after VM call" actions are going to be executed upon returning from

1845

  // ContextBase::*, which might include deleting Context object via proxy_done().

1846

13

  envoyWasm()->addAfterVmCallAction([this, handler] {

1847

13

    http_call_response_ = nullptr;

1848

13

    http_request_.erase(handler);

1849

13

});

1850

13

  ContextBase::onHttpCallResponse(token, response->headers().size(), body_size,

1851

13

                                  headerSize(response->trailers()));

1852

13

1853

1854

6

void Context::onHttpCallFailure(uint32_t token, Http::AsyncClient::FailureReason reason) {

1855

6

  if (proxy_wasm::current_context_ != nullptr) {

1856

    // We are in a reentrant call, so defer.

1857

3

    envoyWasm()->addAfterVmCallAction([this, token, reason] { onHttpCallFailure(token, reason); });

1858

3

    return;

1859

3

1860

3

  auto handler = http_request_.find(token);

1861

3

  if (handler == http_request_.end()) {

1862

    return;

1863

1864

3

  status_code_ = static_cast<uint32_t>(WasmResult::BrokenConnection);

1865

  // TODO(botengyao): handle different failure reasons.

1866

3

  ASSERT(reason == Http::AsyncClient::FailureReason::Reset ||

1867

3

         reason == Http::AsyncClient::FailureReason::ExceedResponseBufferLimit);

1868

3

  status_message_ = "reset";

1869

  // Deferred "after VM call" actions are going to be executed upon returning from

1870

  // ContextBase::*, which might include deleting Context object via proxy_done().

1871

3

  envoyWasm()->addAfterVmCallAction([this, handler] {

1872

3

    status_message_ = "";

1873

3

    http_request_.erase(handler);

1874

3

});

1875

3

  ContextBase::onHttpCallResponse(token, 0, 0, 0);

1876

3

1877

1878

27

void Context::onGrpcReceiveWrapper(uint32_t token, ::Envoy::Buffer::InstancePtr response) {

1879

27

  ASSERT(proxy_wasm::current_context_ == nullptr); // Non-reentrant.

1880

27

  auto cleanup = [this, token] {

1881

27

    if (envoyWasm()->isGrpcCallId(token)) {

1882

7

      grpc_call_request_.erase(token);

1883

7

1884

27

};

1885

27

  if (envoyWasm()->on_grpc_receive_) {

1886

27

    grpc_receive_buffer_ = std::move(response);

1887

27

    uint32_t response_size = grpc_receive_buffer_->length();

1888

    // Deferred "after VM call" actions are going to be executed upon returning from

1889

    // ContextBase::*, which might include deleting Context object via proxy_done().

1890

27

    envoyWasm()->addAfterVmCallAction([this, cleanup] {

1891

27

      grpc_receive_buffer_.reset();

1892

27

      cleanup();

1893

27

});

1894

27

    ContextBase::onGrpcReceive(token, response_size);

1895

27

  } else {

1896

    cleanup();

1897

1898

27

1899

1900

void Context::onGrpcCloseWrapper(uint32_t token, const Grpc::Status::GrpcStatus& status,

1901

22

                                 const std::string_view message) {

1902

22

  if (proxy_wasm::current_context_ != nullptr) {

1903

    // We are in a reentrant call, so defer.

1904

    envoyWasm()->addAfterVmCallAction([this, token, status, message = std::string(message)] {

1905

      onGrpcCloseWrapper(token, status, message);

1906

});

1907

    return;

1908

1909

22

  auto cleanup = [this, token] {

1910

22

    if (envoyWasm()->isGrpcCallId(token)) {

1911

7

      grpc_call_request_.erase(token);

1912

22

    } else if (envoyWasm()->isGrpcStreamId(token)) {

1913

15

      auto it = grpc_stream_.find(token);

1914

15

      if (it != grpc_stream_.end()) {

1915

5

        if (it->second.local_closed_) {

1916

5

          grpc_stream_.erase(token);

1917

5

1918

5

1919

15

1920

22

};

1921

22

  if (envoyWasm()->on_grpc_close_) {

1922

22

    status_code_ = static_cast<uint32_t>(status);

1923

22

    status_message_ = toAbslStringView(message);

1924

    // Deferred "after VM call" actions are going to be executed upon returning from

1925

    // ContextBase::*, which might include deleting Context object via proxy_done().

1926

22

    envoyWasm()->addAfterVmCallAction([this, cleanup] {

1927

22

      status_message_ = "";

1928

22

      cleanup();

1929

22

});

1930

22

    ContextBase::onGrpcClose(token, status_code_);

1931

22

  } else {

1932

    cleanup();

1933

1934

22

1935

1936

30

WasmResult Context::grpcSend(uint32_t token, std::string_view message, bool end_stream) {

1937

30

  if (!envoyWasm()->isGrpcStreamId(token)) {

1938

10

    return WasmResult::BadArgument;

1939

10

1940

20

  auto it = grpc_stream_.find(token);

1941

20

  if (it == grpc_stream_.end()) {

1942

5

    return WasmResult::NotFound;

1943

5

1944

15

  if (it->second.stream_) {

1945

15

    it->second.stream_->sendMessageRaw(::Envoy::Buffer::InstancePtr(new ::Envoy::Buffer::OwnedImpl(

1946

15

                                           message.data(), message.size())),

1947

15

                                       end_stream);

1948

15

1949

15

  return WasmResult::Ok;

1950

20

1951

1952

29

WasmResult Context::grpcClose(uint32_t token) {

1953

29

  if (envoyWasm()->isGrpcCallId(token)) {

1954

9

    auto it = grpc_call_request_.find(token);

1955

9

    if (it == grpc_call_request_.end()) {

1956

5

      return WasmResult::NotFound;

1957

5

1958

4

    if (it->second.request_) {

1959

4

      it->second.request_->cancel();

1960

4

1961

4

    grpc_call_request_.erase(token);

1962

4

    return WasmResult::Ok;

1963

24

  } else if (envoyWasm()->isGrpcStreamId(token)) {

1964

15

    auto it = grpc_stream_.find(token);

1965

15

    if (it == grpc_stream_.end()) {

1966

5

      return WasmResult::NotFound;

1967

5

1968

10

    if (it->second.stream_) {

1969

5

      it->second.stream_->closeStream();

1970

5

1971

10

    if (it->second.remote_closed_) {

1972

5

      grpc_stream_.erase(token);

1973

10

    } else {

1974

5

      it->second.local_closed_ = true;

1975

5

1976

10

    return WasmResult::Ok;

1977

15

1978

5

  return WasmResult::BadArgument;

1979

29

1980

1981

35

WasmResult Context::grpcCancel(uint32_t token) {

1982

35

  if (envoyWasm()->isGrpcCallId(token)) {

1983

15

    auto it = grpc_call_request_.find(token);

1984

15

    if (it == grpc_call_request_.end()) {

1985

5

      return WasmResult::NotFound;

1986

5

1987

10

    if (it->second.request_) {

1988

6

      it->second.request_->cancel();

1989

6

1990

10

    grpc_call_request_.erase(token);

1991

10

    return WasmResult::Ok;

1992

30

  } else if (envoyWasm()->isGrpcStreamId(token)) {

1993

15

    auto it = grpc_stream_.find(token);

1994

15

    if (it == grpc_stream_.end()) {

1995

5

      return WasmResult::NotFound;

1996

5

1997

10

    if (it->second.stream_) {

1998

5

      it->second.stream_->resetStream();

1999

5

2000

10

    grpc_stream_.erase(token);

2001

10

    return WasmResult::Ok;

2002

15

2003

5

  return WasmResult::BadArgument;

2004

35

2005

2006

} // namespace Wasm

2007

} // namespace Common

2008

} // namespace Extensions

2009

} // namespace Envoy