1 /*
2  * Stream processing offload engine management.
3  *
4  * Copyright 2016 HAProxy Technologies, Christopher Faulet <cfaulet@haproxy.com>
5  *
6  * This program is free software; you can redistribute it and/or
7  * modify it under the terms of the GNU General Public License
8  * as published by the Free Software Foundation; either version
9  * 2 of the License, or (at your option) any later version.
10  *
11  */
12 #include <ctype.h>
13 #include <errno.h>
14 
15 #include <common/cfgparse.h>
16 #include <common/compat.h>
17 #include <common/config.h>
18 #include <common/debug.h>
19 #include <common/hathreads.h>
20 #include <common/initcall.h>
21 #include <common/memory.h>
22 #include <common/time.h>
23 
24 #include <types/arg.h>
25 #include <types/global.h>
26 #include <types/spoe.h>
27 
28 #include <proto/acl.h>
29 #include <proto/action.h>
30 #include <proto/arg.h>
31 #include <proto/backend.h>
32 #include <proto/filters.h>
33 #include <proto/freq_ctr.h>
34 #include <proto/frontend.h>
35 #include <proto/http_rules.h>
36 #include <proto/log.h>
37 #include <proto/http_ana.h>
38 #include <proto/proxy.h>
39 #include <proto/sample.h>
40 #include <proto/session.h>
41 #include <proto/signal.h>
42 #include <proto/spoe.h>
43 #include <proto/stream.h>
44 #include <proto/stream_interface.h>
45 #include <proto/task.h>
46 #include <proto/tcp_rules.h>
47 #include <proto/vars.h>
48 
49 #if defined(DEBUG_SPOE) || defined(DEBUG_FULL)
50 #define SPOE_PRINTF(x...) fprintf(x)
51 #define SPOE_DEBUG_STMT(statement) statement
52 #else
53 #define SPOE_PRINTF(x...)
54 #define SPOE_DEBUG_STMT(statement)
55 #endif
56 
57 /* Reserved 4 bytes to the frame size. So a frame and its size can be written
58  * together in a buffer */
59 #define MAX_FRAME_SIZE     global.tune.bufsize - 4
60 
61 /* The minimum size for a frame */
62 #define MIN_FRAME_SIZE     256
63 
64 /* Reserved for the metadata and the frame type.
65  * So <MAX_FRAME_SIZE> - <FRAME_HDR_SIZE> is the maximum payload size */
66 #define FRAME_HDR_SIZE     32
67 
68 /* Helper to get SPOE ctx inside an appctx */
69 #define SPOE_APPCTX(appctx) ((struct spoe_appctx *)((appctx)->ctx.spoe.ptr))
70 
71 /* SPOE filter id. Used to identify SPOE filters */
72 const char *spoe_filter_id = "SPOE filter";
73 
74 /* Set if the handle on SIGUSR1 is registered */
75 static int sighandler_registered = 0;
76 
77 /* proxy used during the parsing */
78 struct proxy *curproxy = NULL;
79 
80 /* The name of the SPOE engine, used during the parsing */
81 char *curengine = NULL;
82 
83 /* SPOE agent used during the parsing */
84 /* SPOE agent/group/message used during the parsing */
85 struct spoe_agent   *curagent = NULL;
86 struct spoe_group   *curgrp   = NULL;
87 struct spoe_message *curmsg   = NULL;
88 
89 /* list of SPOE messages and placeholders used during the parsing */
90 struct list curmsgs;
91 struct list curgrps;
92 struct list curmphs;
93 struct list curgphs;
94 struct list curvars;
95 
96 /* list of log servers used during the parsing */
97 struct list curlogsrvs;
98 
99 /* agent's proxy flags (PR_O_* and PR_O2_*) used during parsing */
100 int curpxopts;
101 int curpxopts2;
102 
103 /* Pools used to allocate SPOE structs */
104 DECLARE_STATIC_POOL(pool_head_spoe_ctx,    "spoe_ctx",    sizeof(struct spoe_context));
105 DECLARE_STATIC_POOL(pool_head_spoe_appctx, "spoe_appctx", sizeof(struct spoe_appctx));
106 
107 struct flt_ops spoe_ops;
108 
109 static int  spoe_queue_context(struct spoe_context *ctx);
110 static int  spoe_acquire_buffer(struct buffer *buf, struct buffer_wait *buffer_wait);
111 static void spoe_release_buffer(struct buffer *buf, struct buffer_wait *buffer_wait);
112 
113 /********************************************************************
114  * helper functions/globals
115  ********************************************************************/
116 static void
spoe_release_placeholder(struct spoe_placeholder * ph)117 spoe_release_placeholder(struct spoe_placeholder *ph)
118 {
119 	if (!ph)
120 		return;
121 	free(ph->id);
122 	free(ph);
123 }
124 
125 static void
spoe_release_message(struct spoe_message * msg)126 spoe_release_message(struct spoe_message *msg)
127 {
128 	struct spoe_arg *arg, *argback;
129 	struct acl      *acl, *aclback;
130 
131 	if (!msg)
132 		return;
133 	free(msg->id);
134 	free(msg->conf.file);
135 	list_for_each_entry_safe(arg, argback, &msg->args, list) {
136 		release_sample_expr(arg->expr);
137 		free(arg->name);
138 		LIST_DEL(&arg->list);
139 		free(arg);
140 	}
141 	list_for_each_entry_safe(acl, aclback, &msg->acls, list) {
142 		LIST_DEL(&acl->list);
143 		prune_acl(acl);
144 		free(acl);
145 	}
146 	if (msg->cond) {
147 		prune_acl_cond(msg->cond);
148 		free(msg->cond);
149 	}
150 	free(msg);
151 }
152 
153 static void
spoe_release_group(struct spoe_group * grp)154 spoe_release_group(struct spoe_group *grp)
155 {
156 	if (!grp)
157 		return;
158 	free(grp->id);
159 	free(grp->conf.file);
160 	free(grp);
161 }
162 
163 static void
spoe_release_agent(struct spoe_agent * agent)164 spoe_release_agent(struct spoe_agent *agent)
165 {
166 	struct spoe_message *msg, *msgback;
167 	struct spoe_group   *grp, *grpback;
168 	int                  i;
169 
170 	if (!agent)
171 		return;
172 	free(agent->id);
173 	free(agent->conf.file);
174 	free(agent->var_pfx);
175 	free(agent->var_on_error);
176 	free(agent->var_t_process);
177 	free(agent->var_t_total);
178 	list_for_each_entry_safe(msg, msgback, &agent->messages, list) {
179 		LIST_DEL(&msg->list);
180 		spoe_release_message(msg);
181 	}
182 	list_for_each_entry_safe(grp, grpback, &agent->groups, list) {
183 		LIST_DEL(&grp->list);
184 		spoe_release_group(grp);
185 	}
186 	if (agent->rt) {
187 		for (i = 0; i < global.nbthread; ++i) {
188 			free(agent->rt[i].engine_id);
189 			HA_SPIN_DESTROY(&agent->rt[i].lock);
190 		}
191 	}
192 	free(agent->rt);
193 	free(agent);
194 }
195 
196 static const char *spoe_frm_err_reasons[SPOE_FRM_ERRS] = {
197 	[SPOE_FRM_ERR_NONE]               = "normal",
198 	[SPOE_FRM_ERR_IO]                 = "I/O error",
199 	[SPOE_FRM_ERR_TOUT]               = "a timeout occurred",
200 	[SPOE_FRM_ERR_TOO_BIG]            = "frame is too big",
201 	[SPOE_FRM_ERR_INVALID]            = "invalid frame received",
202 	[SPOE_FRM_ERR_NO_VSN]             = "version value not found",
203 	[SPOE_FRM_ERR_NO_FRAME_SIZE]      = "max-frame-size value not found",
204 	[SPOE_FRM_ERR_NO_CAP]             = "capabilities value not found",
205 	[SPOE_FRM_ERR_BAD_VSN]            = "unsupported version",
206 	[SPOE_FRM_ERR_BAD_FRAME_SIZE]     = "max-frame-size too big or too small",
207 	[SPOE_FRM_ERR_FRAG_NOT_SUPPORTED] = "fragmentation not supported",
208 	[SPOE_FRM_ERR_INTERLACED_FRAMES]  = "invalid interlaced frames",
209 	[SPOE_FRM_ERR_FRAMEID_NOTFOUND]   = "frame-id not found",
210 	[SPOE_FRM_ERR_RES]                = "resource allocation error",
211 	[SPOE_FRM_ERR_UNKNOWN]            = "an unknown error occurred",
212 };
213 
214 static const char *spoe_event_str[SPOE_EV_EVENTS] = {
215 	[SPOE_EV_ON_CLIENT_SESS] = "on-client-session",
216 	[SPOE_EV_ON_TCP_REQ_FE]  = "on-frontend-tcp-request",
217 	[SPOE_EV_ON_TCP_REQ_BE]  = "on-backend-tcp-request",
218 	[SPOE_EV_ON_HTTP_REQ_FE] = "on-frontend-http-request",
219 	[SPOE_EV_ON_HTTP_REQ_BE] = "on-backend-http-request",
220 
221 	[SPOE_EV_ON_SERVER_SESS] = "on-server-session",
222 	[SPOE_EV_ON_TCP_RSP]     = "on-tcp-response",
223 	[SPOE_EV_ON_HTTP_RSP]    = "on-http-response",
224 };
225 
226 
227 #if defined(DEBUG_SPOE) || defined(DEBUG_FULL)
228 
229 static const char *spoe_ctx_state_str[SPOE_CTX_ST_ERROR+1] = {
230 	[SPOE_CTX_ST_NONE]          = "NONE",
231 	[SPOE_CTX_ST_READY]         = "READY",
232 	[SPOE_CTX_ST_ENCODING_MSGS] = "ENCODING_MSGS",
233 	[SPOE_CTX_ST_SENDING_MSGS]  = "SENDING_MSGS",
234 	[SPOE_CTX_ST_WAITING_ACK]   = "WAITING_ACK",
235 	[SPOE_CTX_ST_DONE]          = "DONE",
236 	[SPOE_CTX_ST_ERROR]         = "ERROR",
237 };
238 
239 static const char *spoe_appctx_state_str[SPOE_APPCTX_ST_END+1] = {
240 	[SPOE_APPCTX_ST_CONNECT]             = "CONNECT",
241 	[SPOE_APPCTX_ST_CONNECTING]          = "CONNECTING",
242 	[SPOE_APPCTX_ST_IDLE]                = "IDLE",
243 	[SPOE_APPCTX_ST_PROCESSING]          = "PROCESSING",
244 	[SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY] = "SENDING_FRAG_NOTIFY",
245 	[SPOE_APPCTX_ST_WAITING_SYNC_ACK]    = "WAITING_SYNC_ACK",
246 	[SPOE_APPCTX_ST_DISCONNECT]          = "DISCONNECT",
247 	[SPOE_APPCTX_ST_DISCONNECTING]       = "DISCONNECTING",
248 	[SPOE_APPCTX_ST_EXIT]                = "EXIT",
249 	[SPOE_APPCTX_ST_END]                 = "END",
250 };
251 
252 #endif
253 
254 /* Used to generates a unique id for an engine. On success, it returns a
255  * allocated string. So it is the caller's reponsibility to release it. If the
256  * allocation failed, it returns NULL. */
257 static char *
generate_pseudo_uuid()258 generate_pseudo_uuid()
259 {
260 	char *uuid;
261 	uint32_t rnd[4] = { 0, 0, 0, 0 };
262 	uint64_t last = 0;
263 	int byte = 0;
264 	uint8_t bits = 0;
265 	unsigned int rand_max_bits = my_flsl(RAND_MAX);
266 
267 	if ((uuid = calloc(1, 37)) == NULL)
268 		return NULL;
269 
270 	while (byte < 4) {
271 		while (bits < 32) {
272 			last |= (uint64_t)ha_random() << bits;
273 			bits += rand_max_bits;
274 		}
275 		rnd[byte++] = last;
276 		last >>= 32u;
277 		bits  -= 32;
278 	}
279 	snprintf(uuid, 37, "%8.8x-%4.4x-%4.4x-%4.4x-%12.12llx",
280 			     rnd[0],
281 			     rnd[1] & 0xFFFF,
282 			     ((rnd[1] >> 16u) & 0xFFF) | 0x4000,  // highest 4 bits indicate the uuid version
283 			     (rnd[2] & 0x3FFF) | 0x8000,  // the highest 2 bits indicate the UUID variant (10),
284 			     (long long)((rnd[2] >> 14u) | ((uint64_t) rnd[3] << 18u)) & 0xFFFFFFFFFFFFull
285 			);
286 	return uuid;
287 }
288 
289 
290 static inline void
spoe_update_stat_time(struct timeval * tv,long * t)291 spoe_update_stat_time(struct timeval *tv, long *t)
292 {
293 	if (*t == -1)
294 		*t = tv_ms_elapsed(tv, &now);
295 	else
296 		*t += tv_ms_elapsed(tv, &now);
297 	tv_zero(tv);
298 }
299 
300 /********************************************************************
301  * Functions that encode/decode SPOE frames
302  ********************************************************************/
303 /* Helper to get static string length, excluding the terminating null byte */
304 #define SLEN(str) (sizeof(str)-1)
305 
306 /* Predefined key used in HELLO/DISCONNECT frames */
307 #define SUPPORTED_VERSIONS_KEY     "supported-versions"
308 #define VERSION_KEY                "version"
309 #define MAX_FRAME_SIZE_KEY         "max-frame-size"
310 #define CAPABILITIES_KEY           "capabilities"
311 #define ENGINE_ID_KEY              "engine-id"
312 #define HEALTHCHECK_KEY            "healthcheck"
313 #define STATUS_CODE_KEY            "status-code"
314 #define MSG_KEY                    "message"
315 
316 struct spoe_version {
317 	char *str;
318 	int   min;
319 	int   max;
320 };
321 
322 /* All supported versions */
323 static struct spoe_version supported_versions[] = {
324 	/* 1.0 is now unsupported because of a bug about frame's flags*/
325 	{"2.0", 2000, 2000},
326 	{NULL,  0, 0}
327 };
328 
329 /* Comma-separated list of supported versions */
330 #define SUPPORTED_VERSIONS_VAL  "2.0"
331 
332 /* Convert a string to a SPOE version value. The string must follow the format
333  * "MAJOR.MINOR". It will be concerted into the integer (1000 * MAJOR + MINOR).
334  * If an error occurred, -1 is returned. */
335 static int
spoe_str_to_vsn(const char * str,size_t len)336 spoe_str_to_vsn(const char *str, size_t len)
337 {
338 	const char *p, *end;
339 	int   maj, min, vsn;
340 
341 	p   = str;
342 	end = str+len;
343 	maj = min = 0;
344 	vsn = -1;
345 
346 	/* skip leading spaces */
347 	while (p < end && isspace(*p))
348 		p++;
349 
350 	/* parse Major number, until the '.' */
351 	while (*p != '.') {
352 		if (p >= end || *p < '0' || *p > '9')
353 			goto out;
354 		maj *= 10;
355 		maj += (*p - '0');
356 		p++;
357 	}
358 
359 	/* check Major version */
360 	if (!maj)
361 		goto out;
362 
363 	p++; /* skip the '.' */
364 	if (p >= end || *p < '0' || *p > '9') /* Minor number is missing */
365 		goto out;
366 
367 	/* Parse Minor number */
368 	while (p < end) {
369 		if (*p < '0' || *p > '9')
370 			break;
371 		min *= 10;
372 		min += (*p - '0');
373 		p++;
374 	}
375 
376 	/* check Minor number */
377 	if (min > 999)
378 		goto out;
379 
380 	/* skip trailing spaces */
381 	while (p < end && isspace(*p))
382 		p++;
383 	if (p != end)
384 		goto out;
385 
386 	vsn = maj * 1000 + min;
387   out:
388 	return vsn;
389 }
390 
391 /* Encode the HELLO frame sent by HAProxy to an agent. It returns the number of
392  * encoded bytes in the frame on success, 0 if an encoding error occurred and -1
393  * if a fatal error occurred. */
394 static int
spoe_prepare_hahello_frame(struct appctx * appctx,char * frame,size_t size)395 spoe_prepare_hahello_frame(struct appctx *appctx, char *frame, size_t size)
396 {
397 	struct buffer      *chk;
398 	struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
399 	char              *p, *end;
400 	unsigned int       flags = SPOE_FRM_FL_FIN;
401 	size_t             sz;
402 
403 	p   = frame;
404 	end = frame+size;
405 
406 	/* Set Frame type */
407 	*p++ = SPOE_FRM_T_HAPROXY_HELLO;
408 
409 	/* Set flags */
410 	flags = htonl(flags);
411 	memcpy(p, (char *)&flags, 4);
412 	p += 4;
413 
414 	/* No stream-id and frame-id for HELLO frames */
415 	*p++ = 0; *p++ = 0;
416 
417 	/* There are 3 mandatory items: "supported-versions", "max-frame-size"
418 	 * and "capabilities" */
419 
420 	/* "supported-versions" K/V item */
421 	sz = SLEN(SUPPORTED_VERSIONS_KEY);
422 	if (spoe_encode_buffer(SUPPORTED_VERSIONS_KEY, sz, &p, end) == -1)
423 		goto too_big;
424 
425 	*p++ = SPOE_DATA_T_STR;
426 	sz = SLEN(SUPPORTED_VERSIONS_VAL);
427 	if (spoe_encode_buffer(SUPPORTED_VERSIONS_VAL, sz, &p, end) == -1)
428 		goto too_big;
429 
430 	/* "max-fram-size" K/V item */
431 	sz = SLEN(MAX_FRAME_SIZE_KEY);
432 	if (spoe_encode_buffer(MAX_FRAME_SIZE_KEY, sz, &p, end) == -1)
433 		goto too_big;
434 
435 	*p++ = SPOE_DATA_T_UINT32;
436 	if (encode_varint(SPOE_APPCTX(appctx)->max_frame_size, &p, end) == -1)
437 		goto too_big;
438 
439 	/* "capabilities" K/V item */
440 	sz = SLEN(CAPABILITIES_KEY);
441 	if (spoe_encode_buffer(CAPABILITIES_KEY, sz, &p, end) == -1)
442 		goto too_big;
443 
444 	*p++ = SPOE_DATA_T_STR;
445 	chk = get_trash_chunk();
446 	if (agent != NULL && (agent->flags & SPOE_FL_PIPELINING)) {
447 		memcpy(chk->area, "pipelining", 10);
448 		chk->data += 10;
449 	}
450 	if (agent != NULL && (agent->flags & SPOE_FL_ASYNC)) {
451 		if (chk->data) chk->area[chk->data++] = ',';
452 		memcpy(chk->area+chk->data, "async", 5);
453 		chk->data += 5;
454 	}
455 	if (agent != NULL && (agent->flags & SPOE_FL_RCV_FRAGMENTATION)) {
456 		if (chk->data) chk->area[chk->data++] = ',';
457 		memcpy(chk->area+chk->data, "fragmentation", 13);
458 		chk->data += 13;
459 	}
460 	if (spoe_encode_buffer(chk->area, chk->data, &p, end) == -1)
461 		goto too_big;
462 
463 	/* (optionnal) "engine-id" K/V item, if present */
464 	if (agent != NULL && agent->rt[tid].engine_id != NULL) {
465 		sz = SLEN(ENGINE_ID_KEY);
466 		if (spoe_encode_buffer(ENGINE_ID_KEY, sz, &p, end) == -1)
467 			goto too_big;
468 
469 		*p++ = SPOE_DATA_T_STR;
470 		sz = strlen(agent->rt[tid].engine_id);
471 		if (spoe_encode_buffer(agent->rt[tid].engine_id, sz, &p, end) == -1)
472 			goto too_big;
473 	}
474 
475 	return (p - frame);
476 
477   too_big:
478 	SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
479 	return 0;
480 }
481 
482 /* Encode DISCONNECT frame sent by HAProxy to an agent. It returns the number of
483  * encoded bytes in the frame on success, 0 if an encoding error occurred and -1
484  * if a fatal error occurred.  */
485 static int
spoe_prepare_hadiscon_frame(struct appctx * appctx,char * frame,size_t size)486 spoe_prepare_hadiscon_frame(struct appctx *appctx, char *frame, size_t size)
487 {
488 	const char  *reason;
489 	char        *p, *end;
490 	unsigned int flags = SPOE_FRM_FL_FIN;
491 	size_t       sz;
492 
493 	p   = frame;
494 	end = frame+size;
495 
496 	 /* Set Frame type */
497 	*p++ = SPOE_FRM_T_HAPROXY_DISCON;
498 
499 	/* Set flags */
500 	flags = htonl(flags);
501 	memcpy(p, (char *)&flags, 4);
502 	p += 4;
503 
504 	/* No stream-id and frame-id for DISCONNECT frames */
505 	*p++ = 0; *p++ = 0;
506 
507 	if (SPOE_APPCTX(appctx)->status_code >= SPOE_FRM_ERRS)
508 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_UNKNOWN;
509 
510 	/* There are 2 mandatory items: "status-code" and "message" */
511 
512 	/* "status-code" K/V item */
513 	sz = SLEN(STATUS_CODE_KEY);
514 	if (spoe_encode_buffer(STATUS_CODE_KEY, sz, &p, end) == -1)
515 		goto too_big;
516 
517 	*p++ = SPOE_DATA_T_UINT32;
518 	if (encode_varint(SPOE_APPCTX(appctx)->status_code, &p, end) == -1)
519 		goto too_big;
520 
521 	/* "message" K/V item */
522 	sz = SLEN(MSG_KEY);
523 	if (spoe_encode_buffer(MSG_KEY, sz, &p, end) == -1)
524 		goto too_big;
525 
526 	/*Get the message corresponding to the status code */
527 	reason = spoe_frm_err_reasons[SPOE_APPCTX(appctx)->status_code];
528 
529 	*p++ = SPOE_DATA_T_STR;
530 	sz = strlen(reason);
531 	if (spoe_encode_buffer(reason, sz, &p, end) == -1)
532 		goto too_big;
533 
534 	return (p - frame);
535 
536   too_big:
537 	SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
538 	return 0;
539 }
540 
541 /* Encode the NOTIFY frame sent by HAProxy to an agent. It returns the number of
542  * encoded bytes in the frame on success, 0 if an encoding error occurred and -1
543  * if a fatal error occurred. */
544 static int
spoe_prepare_hanotify_frame(struct appctx * appctx,struct spoe_context * ctx,char * frame,size_t size)545 spoe_prepare_hanotify_frame(struct appctx *appctx, struct spoe_context *ctx,
546 			    char *frame, size_t size)
547 {
548 	char        *p, *end;
549 	unsigned int stream_id, frame_id;
550 	unsigned int flags = SPOE_FRM_FL_FIN;
551 	size_t       sz;
552 
553 	p   = frame;
554 	end = frame+size;
555 
556 	stream_id = ctx->stream_id;
557 	frame_id  = ctx->frame_id;
558 
559 	if (ctx->flags & SPOE_CTX_FL_FRAGMENTED) {
560 		/* The fragmentation is not supported by the applet */
561 		if (!(SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_FRAGMENTATION)) {
562 			SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
563 			return -1;
564 		}
565 		flags = ctx->frag_ctx.flags;
566 	}
567 
568 	/* Set Frame type */
569 	*p++ = SPOE_FRM_T_HAPROXY_NOTIFY;
570 
571 	/* Set flags */
572 	flags = htonl(flags);
573 	memcpy(p, (char *)&flags, 4);
574 	p += 4;
575 
576 	/* Set stream-id and frame-id */
577 	if (encode_varint(stream_id, &p, end) == -1)
578 		goto too_big;
579 	if (encode_varint(frame_id, &p, end) == -1)
580 		goto too_big;
581 
582 	/* Copy encoded messages, if possible */
583 	sz = b_data(&ctx->buffer);
584 	if (p + sz >= end)
585 		goto too_big;
586 	memcpy(p, b_head(&ctx->buffer), sz);
587 	p += sz;
588 
589 	return (p - frame);
590 
591   too_big:
592 	SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
593 	return 0;
594 }
595 
596 /* Encode next part of a fragmented frame sent by HAProxy to an agent. It
597  * returns the number of encoded bytes in the frame on success, 0 if an encoding
598  * error occurred and -1 if a fatal error occurred. */
599 static int
spoe_prepare_hafrag_frame(struct appctx * appctx,struct spoe_context * ctx,char * frame,size_t size)600 spoe_prepare_hafrag_frame(struct appctx *appctx, struct spoe_context *ctx,
601 			  char *frame, size_t size)
602 {
603 	char        *p, *end;
604 	unsigned int stream_id, frame_id;
605 	unsigned int flags;
606 	size_t       sz;
607 
608 	p   = frame;
609 	end = frame+size;
610 
611 	/* <ctx> is null when the stream has aborted the processing of a
612 	 * fragmented frame. In this case, we must notify the corresponding
613 	 * agent using ids stored in <frag_ctx>. */
614 	if (ctx == NULL) {
615 		flags     = (SPOE_FRM_FL_FIN|SPOE_FRM_FL_ABRT);
616 		stream_id = SPOE_APPCTX(appctx)->frag_ctx.cursid;
617 		frame_id  = SPOE_APPCTX(appctx)->frag_ctx.curfid;
618 	}
619 	else {
620 		flags     = ctx->frag_ctx.flags;
621 		stream_id = ctx->stream_id;
622 		frame_id  = ctx->frame_id;
623 	}
624 
625 	/* Set Frame type */
626 	*p++ = SPOE_FRM_T_UNSET;
627 
628 	/* Set flags */
629 	flags = htonl(flags);
630 	memcpy(p, (char *)&flags, 4);
631 	p += 4;
632 
633 	/* Set stream-id and frame-id */
634 	if (encode_varint(stream_id, &p, end) == -1)
635 		goto too_big;
636 	if (encode_varint(frame_id, &p, end) == -1)
637 		goto too_big;
638 
639 	if (ctx == NULL)
640 		goto end;
641 
642 	/* Copy encoded messages, if possible */
643 	sz = b_data(&ctx->buffer);
644 	if (p + sz >= end)
645 		goto too_big;
646 	memcpy(p, b_head(&ctx->buffer), sz);
647 	p += sz;
648 
649   end:
650 	return (p - frame);
651 
652   too_big:
653 	SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
654 	return 0;
655 }
656 
657 /* Decode and process the HELLO frame sent by an agent. It returns the number of
658  * read bytes on success, 0 if a decoding error occurred, and -1 if a fatal
659  * error occurred. */
660 static int
spoe_handle_agenthello_frame(struct appctx * appctx,char * frame,size_t size)661 spoe_handle_agenthello_frame(struct appctx *appctx, char *frame, size_t size)
662 {
663 	struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
664 	char              *p, *end;
665 	int                vsn, max_frame_size;
666 	unsigned int       flags;
667 
668 	p   = frame;
669 	end = frame + size;
670 
671 	/* Check frame type */
672 	if (*p++ != SPOE_FRM_T_AGENT_HELLO) {
673 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
674 		return 0;
675 	}
676 
677 	if (size < 7 /* TYPE + METADATA */) {
678 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
679 		return 0;
680 	}
681 
682 	/* Retrieve flags */
683 	memcpy((char *)&flags, p, 4);
684 	flags = ntohl(flags);
685 	p += 4;
686 
687 	/* Fragmentation is not supported for HELLO frame */
688 	if (!(flags & SPOE_FRM_FL_FIN)) {
689 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
690 		return -1;
691 	}
692 
693 	/* stream-id and frame-id must be cleared */
694 	if (*p != 0 || *(p+1) != 0) {
695 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
696 		return 0;
697 	}
698 	p += 2;
699 
700 	/* There are 3 mandatory items: "version", "max-frame-size" and
701 	 * "capabilities" */
702 
703 	/* Loop on K/V items */
704 	vsn = max_frame_size = flags = 0;
705 	while (p < end) {
706 		char  *str;
707 		uint64_t sz;
708 		int    ret;
709 
710 		/* Decode the item key */
711 		ret = spoe_decode_buffer(&p, end, &str, &sz);
712 		if (ret == -1 || !sz) {
713 			SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
714 			return 0;
715 		}
716 
717 		/* Check "version" K/V item */
718 		if (sz >= strlen(VERSION_KEY) && !memcmp(str, VERSION_KEY, strlen(VERSION_KEY))) {
719 			int i, type = *p++;
720 
721 			/* The value must be a string */
722 			if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR) {
723 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
724 				return 0;
725 			}
726 			if (spoe_decode_buffer(&p, end, &str, &sz) == -1) {
727 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
728 				return 0;
729 			}
730 
731 			vsn = spoe_str_to_vsn(str, sz);
732 			if (vsn == -1) {
733 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_BAD_VSN;
734 				return -1;
735 			}
736 			for (i = 0; supported_versions[i].str != NULL; ++i) {
737 				if (vsn >= supported_versions[i].min &&
738 				    vsn <= supported_versions[i].max)
739 					break;
740 			}
741 			if (supported_versions[i].str == NULL) {
742 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_BAD_VSN;
743 				return -1;
744 			}
745 		}
746 		/* Check "max-frame-size" K/V item */
747 		else if (sz >= strlen(MAX_FRAME_SIZE_KEY) && !memcmp(str, MAX_FRAME_SIZE_KEY, strlen(MAX_FRAME_SIZE_KEY))) {
748 			int type = *p++;
749 
750 			/* The value must be integer */
751 			if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
752 			    (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
753 			    (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
754 			    (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64) {
755 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
756 				return 0;
757 			}
758 			if (decode_varint(&p, end, &sz) == -1) {
759 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
760 				return 0;
761 			}
762 			if (sz < MIN_FRAME_SIZE ||
763 			    sz > SPOE_APPCTX(appctx)->max_frame_size) {
764 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_BAD_FRAME_SIZE;
765 				return -1;
766 			}
767 			max_frame_size = sz;
768 		}
769 		/* Check "capabilities" K/V item */
770 		else if (sz >= strlen(CAPABILITIES_KEY) && !memcmp(str, CAPABILITIES_KEY, strlen(CAPABILITIES_KEY))) {
771 			int type = *p++;
772 
773 			/* The value must be a string */
774 			if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR) {
775 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
776 				return 0;
777 			}
778 			if (spoe_decode_buffer(&p, end, &str, &sz) == -1) {
779 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
780 				return 0;
781 			}
782 
783 			while (sz) {
784 				char *delim;
785 
786 				/* Skip leading spaces */
787 				for (; isspace(*str) && sz; str++, sz--);
788 
789 				if (sz >= 10 && !strncmp(str, "pipelining", 10)) {
790 					str += 10; sz -= 10;
791 					if (!sz || isspace(*str) || *str == ',')
792 						flags |= SPOE_APPCTX_FL_PIPELINING;
793 				}
794 				else if (sz >= 5 && !strncmp(str, "async", 5)) {
795 					str += 5; sz -= 5;
796 					if (!sz || isspace(*str) || *str == ',')
797 						flags |= SPOE_APPCTX_FL_ASYNC;
798 				}
799 				else if (sz >= 13 && !strncmp(str, "fragmentation", 13)) {
800 					str += 13; sz -= 13;
801 					if (!sz || isspace(*str) || *str == ',')
802 						flags |= SPOE_APPCTX_FL_FRAGMENTATION;
803 				}
804 
805 				/* Get the next comma or break */
806 				if (!sz || (delim = memchr(str, ',', sz)) == NULL)
807 					break;
808 				delim++;
809 				sz -= (delim - str);
810 				str = delim;
811 			}
812 		}
813 		else {
814 			/* Silently ignore unknown item */
815 			if (spoe_skip_data(&p, end) == -1) {
816 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
817 				return 0;
818 			}
819 		}
820 	}
821 
822 	/* Final checks */
823 	if (!vsn) {
824 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NO_VSN;
825 		return -1;
826 	}
827 	if (!max_frame_size) {
828 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NO_FRAME_SIZE;
829 		return -1;
830 	}
831 	if (!agent)
832 		flags &= ~(SPOE_APPCTX_FL_PIPELINING|SPOE_APPCTX_FL_ASYNC);
833 	else {
834 		if ((flags & SPOE_APPCTX_FL_PIPELINING) && !(agent->flags & SPOE_FL_PIPELINING))
835 			flags &= ~SPOE_APPCTX_FL_PIPELINING;
836 		if ((flags & SPOE_APPCTX_FL_ASYNC) && !(agent->flags & SPOE_FL_ASYNC))
837 			flags &= ~SPOE_APPCTX_FL_ASYNC;
838 	}
839 
840 	SPOE_APPCTX(appctx)->version        = (unsigned int)vsn;
841 	SPOE_APPCTX(appctx)->max_frame_size = (unsigned int)max_frame_size;
842 	SPOE_APPCTX(appctx)->flags         |= flags;
843 
844 	return (p - frame);
845 }
846 
847 /* Decode DISCONNECT frame sent by an agent. It returns the number of by read
848  * bytes on success, 0 if the frame can be ignored and -1 if an error
849  * occurred. */
850 static int
spoe_handle_agentdiscon_frame(struct appctx * appctx,char * frame,size_t size)851 spoe_handle_agentdiscon_frame(struct appctx *appctx, char *frame, size_t size)
852 {
853 	char        *p, *end;
854 	unsigned int flags;
855 
856 	p   = frame;
857 	end = frame + size;
858 
859 	/* Check frame type */
860 	if (*p++ != SPOE_FRM_T_AGENT_DISCON) {
861 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
862 		return 0;
863 	}
864 
865 	if (size < 7 /* TYPE + METADATA */) {
866 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
867 		return 0;
868 	}
869 
870 	/* Retrieve flags */
871 	memcpy((char *)&flags, p, 4);
872 	flags = ntohl(flags);
873 	p += 4;
874 
875 	/* Fragmentation is not supported for DISCONNECT frame */
876 	if (!(flags & SPOE_FRM_FL_FIN)) {
877 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
878 		return -1;
879 	}
880 
881 	/* stream-id and frame-id must be cleared */
882 	if (*p != 0 || *(p+1) != 0) {
883 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
884 		return 0;
885 	}
886 	p += 2;
887 
888 	/* There are 2 mandatory items: "status-code" and "message" */
889 
890 	/* Loop on K/V items */
891 	while (p < end) {
892 		char  *str;
893 		uint64_t sz;
894 		int    ret;
895 
896 		/* Decode the item key */
897 		ret = spoe_decode_buffer(&p, end, &str, &sz);
898 		if (ret == -1 || !sz) {
899 			SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
900 			return 0;
901 		}
902 
903 		/* Check "status-code" K/V item */
904 		if (sz >= strlen(STATUS_CODE_KEY) && !memcmp(str, STATUS_CODE_KEY, strlen(STATUS_CODE_KEY))) {
905 			int type = *p++;
906 
907 			/* The value must be an integer */
908 			if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT32 &&
909 			    (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_INT64 &&
910 			    (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT32 &&
911 			    (type & SPOE_DATA_T_MASK) != SPOE_DATA_T_UINT64) {
912 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
913 				return 0;
914 			}
915 			if (decode_varint(&p, end, &sz) == -1) {
916 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
917 				return 0;
918 			}
919 			SPOE_APPCTX(appctx)->status_code = sz;
920 		}
921 
922 		/* Check "message" K/V item */
923 		else if (sz >= strlen(MSG_KEY) && !memcmp(str, MSG_KEY, strlen(MSG_KEY))) {
924 			int type = *p++;
925 
926 			/* The value must be a string */
927 			if ((type & SPOE_DATA_T_MASK) != SPOE_DATA_T_STR) {
928 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
929 				return 0;
930 			}
931 			ret = spoe_decode_buffer(&p, end, &str, &sz);
932 			if (ret == -1 || sz > 255) {
933 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
934 				return 0;
935 			}
936 #if defined(DEBUG_SPOE) || defined(DEBUG_FULL)
937 			SPOE_APPCTX(appctx)->reason = str;
938 			SPOE_APPCTX(appctx)->rlen   = sz;
939 #endif
940 		}
941 		else {
942 			/* Silently ignore unknown item */
943 			if (spoe_skip_data(&p, end) == -1) {
944 				SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
945 				return 0;
946 			}
947 		}
948 	}
949 
950 	return (p - frame);
951 }
952 
953 
954 /* Decode ACK frame sent by an agent. It returns the number of read bytes on
955  * success, 0 if the frame can be ignored and -1 if an error occurred. */
956 static int
spoe_handle_agentack_frame(struct appctx * appctx,struct spoe_context ** ctx,char * frame,size_t size)957 spoe_handle_agentack_frame(struct appctx *appctx, struct spoe_context **ctx,
958 			   char *frame, size_t size)
959 {
960 	struct spoe_agent *agent = SPOE_APPCTX(appctx)->agent;
961 	char              *p, *end;
962 	uint64_t           stream_id, frame_id;
963 	int                len;
964 	unsigned int       flags;
965 
966 	p    = frame;
967 	end  = frame + size;
968 	*ctx = NULL;
969 
970 	/* Check frame type */
971 	if (*p++ != SPOE_FRM_T_AGENT_ACK) {
972 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
973 		return 0;
974 	}
975 
976 	if (size < 7 /* TYPE + METADATA */) {
977 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
978 		return 0;
979 	}
980 
981 	/* Retrieve flags */
982 	memcpy((char *)&flags, p, 4);
983 	flags = ntohl(flags);
984 	p += 4;
985 
986 	/* Fragmentation is not supported for now */
987 	if (!(flags & SPOE_FRM_FL_FIN)) {
988 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAG_NOT_SUPPORTED;
989 		return -1;
990 	}
991 
992 	/* Get the stream-id and the frame-id */
993 	if (decode_varint(&p, end, &stream_id) == -1) {
994 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
995 		return 0;
996 	}
997 	if (decode_varint(&p, end, &frame_id) == -1) {
998 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
999 		return 0;
1000 	}
1001 
1002 	/* Try to find the corresponding SPOE context */
1003 	if (SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_ASYNC) {
1004 		list_for_each_entry((*ctx), &agent->rt[tid].waiting_queue, list) {
1005 			if ((*ctx)->stream_id == (unsigned int)stream_id &&
1006 			    (*ctx)->frame_id  == (unsigned int)frame_id)
1007 				goto found;
1008 		}
1009 	}
1010 	else {
1011 		list_for_each_entry((*ctx), &SPOE_APPCTX(appctx)->waiting_queue, list) {
1012 			if ((*ctx)->stream_id == (unsigned int)stream_id &&
1013 			     (*ctx)->frame_id == (unsigned int)frame_id)
1014 				goto found;
1015 		}
1016 	}
1017 
1018 	if (SPOE_APPCTX(appctx)->frag_ctx.ctx &&
1019 	    SPOE_APPCTX(appctx)->frag_ctx.cursid == (unsigned int)stream_id &&
1020 	    SPOE_APPCTX(appctx)->frag_ctx.curfid == (unsigned int)frame_id) {
1021 
1022 		/* ABRT bit is set for an unfinished fragmented frame */
1023 		if (flags & SPOE_FRM_FL_ABRT) {
1024 			*ctx = SPOE_APPCTX(appctx)->frag_ctx.ctx;
1025 			(*ctx)->state = SPOE_CTX_ST_ERROR;
1026 			(*ctx)->status_code = SPOE_CTX_ERR_FRAG_FRAME_ABRT;
1027 			/* Ignore the payload */
1028 			goto end;
1029 		}
1030 		/* TODO: Handle more flags for fragmented frames: RESUME, FINISH... */
1031 		/*       For now, we ignore the ack */
1032 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_INVALID;
1033 		return 0;
1034 	}
1035 
1036 	/* No Stream found, ignore the frame */
1037 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1038 		    " - Ignore ACK frame"
1039 		    " - stream-id=%u - frame-id=%u\n",
1040 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1041 		    __FUNCTION__, appctx,
1042 		    (unsigned int)stream_id, (unsigned int)frame_id);
1043 
1044 	SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_FRAMEID_NOTFOUND;
1045 	if (appctx->st0 == SPOE_APPCTX_ST_WAITING_SYNC_ACK) {
1046 		/* Report an error if we are waiting the ack for another frame,
1047 		 * but not if there is no longer frame waiting for a ack
1048 		 * (timeout)
1049 		 */
1050 		if (!LIST_ISEMPTY(&SPOE_APPCTX(appctx)->waiting_queue) ||
1051 		    SPOE_APPCTX(appctx)->frag_ctx.ctx)
1052 			return -1;
1053 		appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1054 		SPOE_APPCTX(appctx)->cur_fpa = 0;
1055 	}
1056 	return 0;
1057 
1058   found:
1059 	if (!spoe_acquire_buffer(&SPOE_APPCTX(appctx)->buffer,
1060 				 &SPOE_APPCTX(appctx)->buffer_wait)) {
1061 		*ctx = NULL;
1062 		return 1; /* Retry later */
1063 	}
1064 
1065 	/* Copy encoded actions */
1066 	len = (end - p);
1067 	memcpy(b_head(&SPOE_APPCTX(appctx)->buffer), p, len);
1068 	b_set_data(&SPOE_APPCTX(appctx)->buffer, len);
1069 	p += len;
1070 
1071 	/* Transfer the buffer ownership to the SPOE context */
1072 	(*ctx)->buffer = SPOE_APPCTX(appctx)->buffer;
1073 	SPOE_APPCTX(appctx)->buffer = BUF_NULL;
1074 
1075 	(*ctx)->state = SPOE_CTX_ST_DONE;
1076 
1077   end:
1078 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1079 		    " - ACK frame received"
1080 		    " - ctx=%p - stream-id=%u - frame-id=%u - flags=0x%08x\n",
1081 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1082 		    __FUNCTION__, appctx, *ctx, (*ctx)->stream_id,
1083 		    (*ctx)->frame_id, flags);
1084 	return (p - frame);
1085 }
1086 
1087 /* This function is used in cfgparse.c and declared in proto/checks.h. It
1088  * prepare the request to send to agents during a healthcheck. It returns 0 on
1089  * success and -1 if an error occurred. */
1090 int
spoe_prepare_healthcheck_request(char ** req,int * len)1091 spoe_prepare_healthcheck_request(char **req, int *len)
1092 {
1093 	struct appctx      appctx;
1094 	struct spoe_appctx spoe_appctx;
1095 	char  *frame, *end, buf[MAX_FRAME_SIZE+4];
1096 	size_t sz;
1097 	int    ret;
1098 
1099 	memset(&appctx, 0, sizeof(appctx));
1100 	memset(&spoe_appctx, 0, sizeof(spoe_appctx));
1101 	memset(buf, 0, sizeof(buf));
1102 
1103 	appctx.ctx.spoe.ptr = &spoe_appctx;
1104 	SPOE_APPCTX(&appctx)->max_frame_size = MAX_FRAME_SIZE;
1105 
1106 	frame = buf+4; /* Reserved the 4 first bytes for the frame size */
1107 	end   = frame + MAX_FRAME_SIZE;
1108 
1109 	ret = spoe_prepare_hahello_frame(&appctx, frame, MAX_FRAME_SIZE);
1110 	if (ret <= 0)
1111 		return -1;
1112 	frame += ret;
1113 
1114 	/* Add "healthcheck" K/V item */
1115 	sz = SLEN(HEALTHCHECK_KEY);
1116 	if (spoe_encode_buffer(HEALTHCHECK_KEY, sz, &frame, end) == -1)
1117 		return -1;
1118 	*frame++ = (SPOE_DATA_T_BOOL | SPOE_DATA_FL_TRUE);
1119 
1120 	*len = frame - buf;
1121 	sz   = htonl(*len - 4);
1122 	memcpy(buf, (char *)&sz, 4);
1123 
1124 	if ((*req = malloc(*len)) == NULL)
1125 		return -1;
1126 	memcpy(*req, buf, *len);
1127 	return 0;
1128 }
1129 
1130 /* This function is used in checks.c and declared in proto/checks.h. It decode
1131  * the response received from an agent during a healthcheck. It returns 0 on
1132  * success and -1 if an error occurred. */
1133 int
spoe_handle_healthcheck_response(char * frame,size_t size,char * err,int errlen)1134 spoe_handle_healthcheck_response(char *frame, size_t size, char *err, int errlen)
1135 {
1136 	struct appctx      appctx;
1137 	struct spoe_appctx spoe_appctx;
1138 
1139 	memset(&appctx, 0, sizeof(appctx));
1140 	memset(&spoe_appctx, 0, sizeof(spoe_appctx));
1141 
1142 	appctx.ctx.spoe.ptr = &spoe_appctx;
1143 	SPOE_APPCTX(&appctx)->max_frame_size = MAX_FRAME_SIZE;
1144 
1145 	if (*frame == SPOE_FRM_T_AGENT_DISCON) {
1146 		spoe_handle_agentdiscon_frame(&appctx, frame, size);
1147 		goto error;
1148 	}
1149 	if (spoe_handle_agenthello_frame(&appctx, frame, size) <= 0)
1150 		goto error;
1151 
1152 	return 0;
1153 
1154   error:
1155 	if (SPOE_APPCTX(&appctx)->status_code >= SPOE_FRM_ERRS)
1156 		SPOE_APPCTX(&appctx)->status_code = SPOE_FRM_ERR_UNKNOWN;
1157 	strncpy(err, spoe_frm_err_reasons[SPOE_APPCTX(&appctx)->status_code], errlen);
1158 	return -1;
1159 }
1160 
1161 /* Send a SPOE frame to an agent. It returns -1 when an error occurred, 0 when
1162  * the frame can be ignored, 1 to retry later, and the frame legnth on
1163  * success. */
1164 static int
spoe_send_frame(struct appctx * appctx,char * buf,size_t framesz)1165 spoe_send_frame(struct appctx *appctx, char *buf, size_t framesz)
1166 {
1167 	struct stream_interface *si = appctx->owner;
1168 	int      ret;
1169 	uint32_t netint;
1170 
1171 	/* 4 bytes are reserved at the beginning of <buf> to store the frame
1172 	 * length. */
1173 	netint = htonl(framesz);
1174 	memcpy(buf, (char *)&netint, 4);
1175 	ret = ci_putblk(si_ic(si), buf, framesz+4);
1176 	if (ret <= 0) {
1177 		if ((ret == -3 && b_is_null(&si_ic(si)->buf)) || ret == -1) {
1178 			si_rx_room_blk(si);
1179 			return 1; /* retry */
1180 		}
1181 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1182 		return -1; /* error */
1183 	}
1184 	return framesz;
1185 }
1186 
1187 /* Receive a SPOE frame from an agent. It return -1 when an error occurred, 0
1188  * when the frame can be ignored, 1 to retry later and the frame length on
1189  * success. */
1190 static int
spoe_recv_frame(struct appctx * appctx,char * buf,size_t framesz)1191 spoe_recv_frame(struct appctx *appctx, char *buf, size_t framesz)
1192 {
1193 	struct stream_interface *si = appctx->owner;
1194 	int      ret;
1195 	uint32_t netint;
1196 
1197 	ret = co_getblk(si_oc(si), (char *)&netint, 4, 0);
1198 	if (ret > 0) {
1199 		framesz = ntohl(netint);
1200 		if (framesz > SPOE_APPCTX(appctx)->max_frame_size) {
1201 			SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOO_BIG;
1202 			return -1;
1203 		}
1204 		ret = co_getblk(si_oc(si), buf, framesz, 4);
1205 	}
1206 	if (ret <= 0) {
1207 		if (ret == 0) {
1208 			return 1; /* retry */
1209 		}
1210 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1211 		return -1; /* error */
1212 	}
1213 	return framesz;
1214 }
1215 
1216 /********************************************************************
1217  * Functions that manage the SPOE applet
1218  ********************************************************************/
1219 static int
spoe_wakeup_appctx(struct appctx * appctx)1220 spoe_wakeup_appctx(struct appctx *appctx)
1221 {
1222 	si_want_get(appctx->owner);
1223 	si_rx_endp_more(appctx->owner);
1224 	appctx_wakeup(appctx);
1225 	return 1;
1226 }
1227 
1228 /* Callback function that catches applet timeouts. If a timeout occurred, we set
1229  * <appctx->st1> flag and the SPOE applet is woken up. */
1230 static struct task *
spoe_process_appctx(struct task * task,void * context,unsigned short state)1231 spoe_process_appctx(struct task * task, void *context, unsigned short state)
1232 {
1233 	struct appctx *appctx = context;
1234 
1235 	appctx->st1 = SPOE_APPCTX_ERR_NONE;
1236 	if (tick_is_expired(task->expire, now_ms)) {
1237 		task->expire = TICK_ETERNITY;
1238 		appctx->st1  = SPOE_APPCTX_ERR_TOUT;
1239 	}
1240 	spoe_wakeup_appctx(appctx);
1241 	return task;
1242 }
1243 
1244 /* Callback function that releases a SPOE applet. This happens when the
1245  * connection with the agent is closed. */
1246 static void
spoe_release_appctx(struct appctx * appctx)1247 spoe_release_appctx(struct appctx *appctx)
1248 {
1249 	struct stream_interface *si          = appctx->owner;
1250 	struct spoe_appctx      *spoe_appctx = SPOE_APPCTX(appctx);
1251 	struct spoe_agent       *agent;
1252 	struct spoe_context     *ctx, *back;
1253 
1254 	if (spoe_appctx == NULL)
1255 		return;
1256 
1257 	appctx->ctx.spoe.ptr = NULL;
1258 	agent = spoe_appctx->agent;
1259 
1260 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p\n",
1261 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1262 		    __FUNCTION__, appctx);
1263 
1264 	/* Remove applet from the list of running applets */
1265 	_HA_ATOMIC_SUB(&agent->counters.applets, 1);
1266 	HA_SPIN_LOCK(SPOE_APPLET_LOCK, &agent->rt[tid].lock);
1267 	if (!LIST_ISEMPTY(&spoe_appctx->list)) {
1268 		LIST_DEL(&spoe_appctx->list);
1269 		LIST_INIT(&spoe_appctx->list);
1270 	}
1271 	HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &agent->rt[tid].lock);
1272 
1273 	/* Shutdown the server connection, if needed */
1274 	if (appctx->st0 != SPOE_APPCTX_ST_END) {
1275 		if (appctx->st0 == SPOE_APPCTX_ST_IDLE) {
1276 			eb32_delete(&spoe_appctx->node);
1277 			_HA_ATOMIC_SUB(&agent->counters.idles, 1);
1278 		}
1279 
1280 		appctx->st0 = SPOE_APPCTX_ST_END;
1281 		if (spoe_appctx->status_code == SPOE_FRM_ERR_NONE)
1282 			spoe_appctx->status_code = SPOE_FRM_ERR_IO;
1283 
1284 		si_shutw(si);
1285 		si_shutr(si);
1286 		si_ic(si)->flags |= CF_READ_NULL;
1287 	}
1288 
1289 	/* Destroy the task attached to this applet */
1290 	task_destroy(spoe_appctx->task);
1291 
1292 	/* Notify all waiting streams */
1293 	list_for_each_entry_safe(ctx, back, &spoe_appctx->waiting_queue, list) {
1294 		LIST_DEL(&ctx->list);
1295 		LIST_INIT(&ctx->list);
1296 		_HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
1297 		spoe_update_stat_time(&ctx->stats.tv_wait, &ctx->stats.t_waiting);
1298 		ctx->spoe_appctx = NULL;
1299 		ctx->state = SPOE_CTX_ST_ERROR;
1300 		ctx->status_code = (spoe_appctx->status_code + 0x100);
1301 		task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1302 	}
1303 
1304 	/* If the applet was processing a fragmented frame, notify the
1305 	 * corresponding stream. */
1306 	if (spoe_appctx->frag_ctx.ctx) {
1307 		ctx = spoe_appctx->frag_ctx.ctx;
1308 		ctx->spoe_appctx = NULL;
1309 		ctx->state = SPOE_CTX_ST_ERROR;
1310 		ctx->status_code = (spoe_appctx->status_code + 0x100);
1311 		task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1312 	}
1313 
1314 	if (!LIST_ISEMPTY(&agent->rt[tid].applets)) {
1315 		list_for_each_entry_safe(ctx, back, &agent->rt[tid].waiting_queue, list) {
1316 			if (ctx->spoe_appctx == spoe_appctx)
1317 				ctx->spoe_appctx = NULL;
1318 		}
1319 		goto end;
1320 	}
1321 
1322 	/* If this was the last running applet, notify all waiting streams */
1323 	list_for_each_entry_safe(ctx, back, &agent->rt[tid].sending_queue, list) {
1324 		LIST_DEL(&ctx->list);
1325 		LIST_INIT(&ctx->list);
1326 		_HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
1327 		spoe_update_stat_time(&ctx->stats.tv_queue, &ctx->stats.t_queue);
1328 		ctx->spoe_appctx = NULL;
1329 		ctx->state = SPOE_CTX_ST_ERROR;
1330 		ctx->status_code = (spoe_appctx->status_code + 0x100);
1331 		task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1332 	}
1333 	list_for_each_entry_safe(ctx, back, &agent->rt[tid].waiting_queue, list) {
1334 		LIST_DEL(&ctx->list);
1335 		LIST_INIT(&ctx->list);
1336 		_HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
1337 		spoe_update_stat_time(&ctx->stats.tv_wait, &ctx->stats.t_waiting);
1338 		ctx->spoe_appctx = NULL;
1339 		ctx->state = SPOE_CTX_ST_ERROR;
1340 		ctx->status_code = (spoe_appctx->status_code + 0x100);
1341 		task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1342 	}
1343 
1344   end:
1345 	/* Release allocated memory */
1346 	spoe_release_buffer(&spoe_appctx->buffer,
1347 			    &spoe_appctx->buffer_wait);
1348 	pool_free(pool_head_spoe_appctx, spoe_appctx);
1349 
1350 	/* Update runtinme agent info */
1351 	agent->rt[tid].frame_size = agent->max_frame_size;
1352 	list_for_each_entry(spoe_appctx, &agent->rt[tid].applets, list)
1353 		HA_ATOMIC_UPDATE_MIN(&agent->rt[tid].frame_size, spoe_appctx->max_frame_size);
1354 }
1355 
1356 static int
spoe_handle_connect_appctx(struct appctx * appctx)1357 spoe_handle_connect_appctx(struct appctx *appctx)
1358 {
1359 	struct stream_interface *si    = appctx->owner;
1360 	struct spoe_agent       *agent = SPOE_APPCTX(appctx)->agent;
1361 	char *frame, *buf;
1362 	int   ret;
1363 
1364 	if (si_state_in(si->state, SI_SB_CER|SI_SB_DIS|SI_SB_CLO)) {
1365 		/* closed */
1366 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1367 		goto exit;
1368 	}
1369 
1370 	if (!si_state_in(si->state, SI_SB_RDY|SI_SB_EST)) {
1371 		/* not connected yet */
1372 		si_rx_endp_more(si);
1373 		task_wakeup(si_strm(si)->task, TASK_WOKEN_MSG);
1374 		goto stop;
1375 	}
1376 
1377 	if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1378 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1379 			    " - Connection timed out\n",
1380 			    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1381 			    __FUNCTION__, appctx);
1382 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1383 		goto exit;
1384 	}
1385 
1386 	if (SPOE_APPCTX(appctx)->task->expire == TICK_ETERNITY)
1387 		SPOE_APPCTX(appctx)->task->expire =
1388 			tick_add_ifset(now_ms, agent->timeout.hello);
1389 
1390 	/* 4 bytes are reserved at the beginning of <buf> to store the frame
1391 	 * length. */
1392 	buf = trash.area; frame = buf+4;
1393 	ret = spoe_prepare_hahello_frame(appctx, frame,
1394 					 SPOE_APPCTX(appctx)->max_frame_size);
1395 	if (ret > 1)
1396 		ret = spoe_send_frame(appctx, buf, ret);
1397 
1398 	switch (ret) {
1399 		case -1: /* error */
1400 		case  0: /* ignore => an error, cannot be ignored */
1401 			goto exit;
1402 
1403 		case  1: /* retry later */
1404 			goto stop;
1405 
1406 		default:
1407 			/* HELLO frame successfully sent, now wait for the
1408 			 * reply. */
1409 			appctx->st0 = SPOE_APPCTX_ST_CONNECTING;
1410 			goto next;
1411 	}
1412 
1413   next:
1414 	return 0;
1415   stop:
1416 	return 1;
1417   exit:
1418 	appctx->st0 = SPOE_APPCTX_ST_EXIT;
1419 	return 0;
1420 }
1421 
1422 static int
spoe_handle_connecting_appctx(struct appctx * appctx)1423 spoe_handle_connecting_appctx(struct appctx *appctx)
1424 {
1425 	struct stream_interface *si     = appctx->owner;
1426 	struct spoe_agent       *agent  = SPOE_APPCTX(appctx)->agent;
1427 	char  *frame;
1428 	int    ret;
1429 
1430 
1431 	if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
1432 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1433 		goto exit;
1434 	}
1435 
1436 	if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1437 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1438 			    " - Connection timed out\n",
1439 			    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1440 			    __FUNCTION__, appctx);
1441 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1442 		goto exit;
1443 	}
1444 
1445 	frame = trash.area; trash.data = 0;
1446 	ret = spoe_recv_frame(appctx, frame,
1447 			      SPOE_APPCTX(appctx)->max_frame_size);
1448 	if (ret > 1) {
1449 		if (*frame == SPOE_FRM_T_AGENT_DISCON) {
1450 			appctx->st0 = SPOE_APPCTX_ST_DISCONNECTING;
1451 			goto next;
1452 		}
1453 		trash.data = ret + 4;
1454 		ret = spoe_handle_agenthello_frame(appctx, frame, ret);
1455 	}
1456 
1457 	switch (ret) {
1458 		case -1: /* error */
1459 		case  0: /* ignore => an error, cannot be ignored */
1460 			appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1461 			goto next;
1462 
1463 		case 1: /* retry later */
1464 			goto stop;
1465 
1466 		default:
1467 			_HA_ATOMIC_ADD(&agent->counters.idles, 1);
1468 			appctx->st0 = SPOE_APPCTX_ST_IDLE;
1469 			SPOE_APPCTX(appctx)->node.key = 0;
1470 			eb32_insert(&agent->rt[tid].idle_applets, &SPOE_APPCTX(appctx)->node);
1471 
1472 			/* Update runtinme agent info */
1473 			HA_ATOMIC_UPDATE_MIN(&agent->rt[tid].frame_size, SPOE_APPCTX(appctx)->max_frame_size);
1474 			goto next;
1475 	}
1476 
1477   next:
1478 	/* Do not forget to remove processed frame from the output buffer */
1479 	if (trash.data)
1480 		co_skip(si_oc(si), trash.data);
1481 
1482 	SPOE_APPCTX(appctx)->task->expire =
1483 		tick_add_ifset(now_ms, agent->timeout.idle);
1484 	return 0;
1485   stop:
1486 	return 1;
1487   exit:
1488 	appctx->st0 = SPOE_APPCTX_ST_EXIT;
1489 	return 0;
1490 }
1491 
1492 
1493 static int
spoe_handle_sending_frame_appctx(struct appctx * appctx,int * skip)1494 spoe_handle_sending_frame_appctx(struct appctx *appctx, int *skip)
1495 {
1496 	struct spoe_agent   *agent = SPOE_APPCTX(appctx)->agent;
1497 	struct spoe_context *ctx = NULL;
1498 	char *frame, *buf;
1499 	int   ret;
1500 
1501 	/* 4 bytes are reserved at the beginning of <buf> to store the frame
1502 	 * length. */
1503 	buf = trash.area; frame = buf+4;
1504 
1505 	if (appctx->st0 == SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY) {
1506 		ctx = SPOE_APPCTX(appctx)->frag_ctx.ctx;
1507 		ret = spoe_prepare_hafrag_frame(appctx, ctx, frame,
1508 						SPOE_APPCTX(appctx)->max_frame_size);
1509 	}
1510 	else if (LIST_ISEMPTY(&agent->rt[tid].sending_queue)) {
1511 		*skip = 1;
1512 		ret   = 1;
1513 		goto end;
1514 	}
1515 	else {
1516 		ctx = LIST_NEXT(&agent->rt[tid].sending_queue, typeof(ctx), list);
1517 		ret = spoe_prepare_hanotify_frame(appctx, ctx, frame,
1518 						  SPOE_APPCTX(appctx)->max_frame_size);
1519 
1520 	}
1521 
1522 	if (ret > 1)
1523 		ret = spoe_send_frame(appctx, buf, ret);
1524 
1525 	switch (ret) {
1526 		case -1: /* error */
1527 			appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1528 			goto end;
1529 
1530 		case 0: /* ignore */
1531 			if (ctx == NULL)
1532 				goto abort_frag_frame;
1533 
1534 			spoe_release_buffer(&ctx->buffer, &ctx->buffer_wait);
1535 			LIST_DEL(&ctx->list);
1536 			LIST_INIT(&ctx->list);
1537 			_HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
1538 			spoe_update_stat_time(&ctx->stats.tv_queue, &ctx->stats.t_queue);
1539 			ctx->spoe_appctx = NULL;
1540 			ctx->state = SPOE_CTX_ST_ERROR;
1541 			ctx->status_code = (SPOE_APPCTX(appctx)->status_code + 0x100);
1542 			task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1543 			*skip = 1;
1544 			break;
1545 
1546 		case 1: /* retry */
1547 			*skip = 1;
1548 			break;
1549 
1550 		default:
1551 			if (ctx == NULL)
1552 				goto abort_frag_frame;
1553 
1554 			spoe_release_buffer(&ctx->buffer, &ctx->buffer_wait);
1555 			LIST_DEL(&ctx->list);
1556 			LIST_INIT(&ctx->list);
1557 			_HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
1558 			spoe_update_stat_time(&ctx->stats.tv_queue, &ctx->stats.t_queue);
1559 			ctx->spoe_appctx = SPOE_APPCTX(appctx);
1560 			if (!(ctx->flags & SPOE_CTX_FL_FRAGMENTED) ||
1561 			    (ctx->frag_ctx.flags & SPOE_FRM_FL_FIN))
1562 				goto no_frag_frame_sent;
1563 			else
1564 				goto frag_frame_sent;
1565 	}
1566 	goto end;
1567 
1568   frag_frame_sent:
1569 	appctx->st0 = SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY;
1570 	*skip = 1;
1571 	SPOE_APPCTX(appctx)->frag_ctx.ctx    = ctx;
1572 	SPOE_APPCTX(appctx)->frag_ctx.cursid = ctx->stream_id;
1573 	SPOE_APPCTX(appctx)->frag_ctx.curfid = ctx->frame_id;
1574 	ctx->state = SPOE_CTX_ST_ENCODING_MSGS;
1575 	task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1576 	goto end;
1577 
1578   no_frag_frame_sent:
1579 	if (SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_ASYNC) {
1580 		appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1581 		LIST_ADDQ(&agent->rt[tid].waiting_queue, &ctx->list);
1582 	}
1583 	else if (SPOE_APPCTX(appctx)->flags & SPOE_APPCTX_FL_PIPELINING) {
1584 		appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1585 		LIST_ADDQ(&SPOE_APPCTX(appctx)->waiting_queue, &ctx->list);
1586 	}
1587 	else {
1588 		appctx->st0 = SPOE_APPCTX_ST_WAITING_SYNC_ACK;
1589 		*skip = 1;
1590 		LIST_ADDQ(&SPOE_APPCTX(appctx)->waiting_queue, &ctx->list);
1591 	}
1592 	_HA_ATOMIC_ADD(&agent->counters.nb_waiting, 1);
1593 	ctx->stats.tv_wait = now;
1594 	SPOE_APPCTX(appctx)->frag_ctx.ctx    = NULL;
1595 	SPOE_APPCTX(appctx)->frag_ctx.cursid = 0;
1596 	SPOE_APPCTX(appctx)->frag_ctx.curfid = 0;
1597 	SPOE_APPCTX(appctx)->cur_fpa++;
1598 
1599 	ctx->state = SPOE_CTX_ST_WAITING_ACK;
1600 	goto end;
1601 
1602   abort_frag_frame:
1603 	appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1604 	SPOE_APPCTX(appctx)->frag_ctx.ctx    = NULL;
1605 	SPOE_APPCTX(appctx)->frag_ctx.cursid = 0;
1606 	SPOE_APPCTX(appctx)->frag_ctx.curfid = 0;
1607 	goto end;
1608 
1609   end:
1610 	return ret;
1611 }
1612 
1613 static int
spoe_handle_receiving_frame_appctx(struct appctx * appctx,int * skip)1614 spoe_handle_receiving_frame_appctx(struct appctx *appctx, int *skip)
1615 {
1616 	struct spoe_agent   *agent = SPOE_APPCTX(appctx)->agent;
1617 	struct spoe_context *ctx = NULL;
1618 	char *frame;
1619 	int   ret;
1620 
1621 	frame = trash.area; trash.data = 0;
1622 	ret = spoe_recv_frame(appctx, frame,
1623 			      SPOE_APPCTX(appctx)->max_frame_size);
1624 	if (ret > 1) {
1625 		if (*frame == SPOE_FRM_T_AGENT_DISCON) {
1626 			appctx->st0 = SPOE_APPCTX_ST_DISCONNECTING;
1627 			ret = -1;
1628 			goto end;
1629 		}
1630 		trash.data = ret + 4;
1631 		ret = spoe_handle_agentack_frame(appctx, &ctx, frame, ret);
1632 	}
1633 	switch (ret) {
1634 		case -1: /* error */
1635 			appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1636 			break;
1637 
1638 		case 0: /* ignore */
1639 			break;
1640 
1641 		case 1: /* retry */
1642 			*skip = 1;
1643 			break;
1644 
1645 		default:
1646 			LIST_DEL(&ctx->list);
1647 			LIST_INIT(&ctx->list);
1648 			_HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
1649 			spoe_update_stat_time(&ctx->stats.tv_wait, &ctx->stats.t_waiting);
1650 			ctx->stats.tv_response = now;
1651 			if (ctx->spoe_appctx) {
1652 				ctx->spoe_appctx->cur_fpa--;
1653 				ctx->spoe_appctx = NULL;
1654 			}
1655 			if (appctx->st0 == SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY &&
1656 			    ctx == SPOE_APPCTX(appctx)->frag_ctx.ctx) {
1657 				appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1658 				SPOE_APPCTX(appctx)->frag_ctx.ctx    = NULL;
1659 				SPOE_APPCTX(appctx)->frag_ctx.cursid = 0;
1660 				SPOE_APPCTX(appctx)->frag_ctx.curfid = 0;
1661 			}
1662 			else if (appctx->st0 == SPOE_APPCTX_ST_WAITING_SYNC_ACK)
1663 				appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1664 			task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
1665 			break;
1666 	}
1667 
1668 	/* Do not forget to remove processed frame from the output buffer */
1669 	if (trash.data)
1670 		co_skip(si_oc(appctx->owner), trash.data);
1671   end:
1672 	return ret;
1673 }
1674 
1675 static int
spoe_handle_processing_appctx(struct appctx * appctx)1676 spoe_handle_processing_appctx(struct appctx *appctx)
1677 {
1678 	struct stream_interface *si    = appctx->owner;
1679 	struct spoe_agent       *agent = SPOE_APPCTX(appctx)->agent;
1680 	int ret, skip_sending = 0, skip_receiving = 0, active_s = 0, active_r = 0;
1681 
1682 	if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
1683 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1684 		goto exit;
1685 	}
1686 
1687 	if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1688 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1689 		appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1690 		appctx->st1 = SPOE_APPCTX_ERR_NONE;
1691 		goto next;
1692 	}
1693 
1694 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1695 		    " - process: fpa=%u/%u - appctx-state=%s - weight=%u - flags=0x%08x\n",
1696 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1697 		    __FUNCTION__, appctx, SPOE_APPCTX(appctx)->cur_fpa,
1698 		    agent->max_fpa, spoe_appctx_state_str[appctx->st0],
1699 		    SPOE_APPCTX(appctx)->node.key, SPOE_APPCTX(appctx)->flags);
1700 
1701 	if (appctx->st0 == SPOE_APPCTX_ST_WAITING_SYNC_ACK)
1702 		skip_sending = 1;
1703 
1704 	/* receiving_frame loop */
1705 	while (!skip_receiving) {
1706 		ret = spoe_handle_receiving_frame_appctx(appctx, &skip_receiving);
1707 		switch (ret) {
1708 			case -1: /* error */
1709 				goto next;
1710 
1711 			case 0: /* ignore */
1712 				active_r = 1;
1713 				break;
1714 
1715 			case 1: /* retry */
1716 				break;
1717 
1718 			default:
1719 				active_r = 1;
1720 				break;
1721 		}
1722 	}
1723 
1724 	/* send_frame loop */
1725 	while (!skip_sending && SPOE_APPCTX(appctx)->cur_fpa < agent->max_fpa) {
1726 		ret = spoe_handle_sending_frame_appctx(appctx, &skip_sending);
1727 		switch (ret) {
1728 			case -1: /* error */
1729 				goto next;
1730 
1731 			case 0: /* ignore */
1732 				if (SPOE_APPCTX(appctx)->node.key)
1733 					SPOE_APPCTX(appctx)->node.key--;
1734 				active_s++;
1735 				break;
1736 
1737 			case 1: /* retry */
1738 				break;
1739 
1740 			default:
1741 				if (SPOE_APPCTX(appctx)->node.key)
1742 					SPOE_APPCTX(appctx)->node.key--;
1743 				active_s++;
1744 				break;
1745 		}
1746 	}
1747 
1748 	if (active_s || active_r) {
1749 		update_freq_ctr(&agent->rt[tid].processing_per_sec, active_s);
1750 		SPOE_APPCTX(appctx)->task->expire = tick_add_ifset(now_ms, agent->timeout.idle);
1751 	}
1752 
1753 	if (appctx->st0 == SPOE_APPCTX_ST_PROCESSING && SPOE_APPCTX(appctx)->cur_fpa < agent->max_fpa) {
1754 		struct server *srv = objt_server(si_strm(si)->target);
1755 
1756 		/* With several threads, close the applet if there are pending
1757 		 * connections or if the server is full. Otherwise, add the
1758 		 * applet in the idle list.
1759 		 */
1760 		if (global.nbthread > 1 &&
1761 		    (agent->b.be->nbpend ||
1762 		     (srv && (srv->nbpend || (srv->maxconn && srv->served >=srv_dynamic_maxconn(srv)))))) {
1763 			SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NONE;
1764 			appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1765 			appctx->st1 = SPOE_APPCTX_ERR_NONE;
1766 			goto next;
1767 		}
1768 		_HA_ATOMIC_ADD(&agent->counters.idles, 1);
1769 		appctx->st0 = SPOE_APPCTX_ST_IDLE;
1770 		eb32_insert(&agent->rt[tid].idle_applets, &SPOE_APPCTX(appctx)->node);
1771 	}
1772 	return 1;
1773 
1774   next:
1775 	SPOE_APPCTX(appctx)->task->expire = tick_add_ifset(now_ms, agent->timeout.idle);
1776 	return 0;
1777 
1778   exit:
1779 	appctx->st0 = SPOE_APPCTX_ST_EXIT;
1780 	return 0;
1781 }
1782 
1783 static int
spoe_handle_disconnect_appctx(struct appctx * appctx)1784 spoe_handle_disconnect_appctx(struct appctx *appctx)
1785 {
1786 	struct stream_interface *si    = appctx->owner;
1787 	struct spoe_agent       *agent = SPOE_APPCTX(appctx)->agent;
1788 	char *frame, *buf;
1789 	int   ret;
1790 
1791 	if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO)
1792 		goto exit;
1793 
1794 	if (appctx->st1 == SPOE_APPCTX_ERR_TOUT)
1795 		goto exit;
1796 
1797 	/* 4 bytes are reserved at the beginning of <buf> to store the frame
1798 	 * length. */
1799 	buf = trash.area; frame = buf+4;
1800 	ret = spoe_prepare_hadiscon_frame(appctx, frame,
1801 					  SPOE_APPCTX(appctx)->max_frame_size);
1802 	if (ret > 1)
1803 		ret = spoe_send_frame(appctx, buf, ret);
1804 
1805 	switch (ret) {
1806 		case -1: /* error */
1807 		case  0: /* ignore  => an error, cannot be ignored */
1808 			goto exit;
1809 
1810 		case 1: /* retry */
1811 			goto stop;
1812 
1813 		default:
1814 			SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1815 				    " - disconnected by HAProxy (%d): %s\n",
1816 				    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1817 				    __FUNCTION__, appctx,
1818 				    SPOE_APPCTX(appctx)->status_code,
1819 				    spoe_frm_err_reasons[SPOE_APPCTX(appctx)->status_code]);
1820 
1821 			appctx->st0 = SPOE_APPCTX_ST_DISCONNECTING;
1822 			goto next;
1823 	}
1824 
1825   next:
1826 	SPOE_APPCTX(appctx)->task->expire =
1827 		tick_add_ifset(now_ms, agent->timeout.idle);
1828 	return 0;
1829   stop:
1830 	return 1;
1831   exit:
1832 	appctx->st0 = SPOE_APPCTX_ST_EXIT;
1833 	return 0;
1834 }
1835 
1836 static int
spoe_handle_disconnecting_appctx(struct appctx * appctx)1837 spoe_handle_disconnecting_appctx(struct appctx *appctx)
1838 {
1839 	struct stream_interface *si = appctx->owner;
1840 	char  *frame;
1841 	int    ret;
1842 
1843 	if (si->state == SI_ST_CLO || si_opposite(si)->state == SI_ST_CLO) {
1844 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_IO;
1845 		goto exit;
1846 	}
1847 
1848 	if (appctx->st1 == SPOE_APPCTX_ERR_TOUT) {
1849 		SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_TOUT;
1850 		goto exit;
1851 	}
1852 
1853 	frame = trash.area; trash.data = 0;
1854 	ret = spoe_recv_frame(appctx, frame,
1855 			      SPOE_APPCTX(appctx)->max_frame_size);
1856 	if (ret > 1) {
1857 		trash.data = ret + 4;
1858 		ret = spoe_handle_agentdiscon_frame(appctx, frame, ret);
1859 	}
1860 
1861 	switch (ret) {
1862 		case -1: /* error  */
1863 			SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1864 				    " - error on frame (%s)\n",
1865 				    (int)now.tv_sec, (int)now.tv_usec,
1866 				    ((struct spoe_agent *)SPOE_APPCTX(appctx)->agent)->id,
1867 				    __FUNCTION__, appctx,
1868 				    spoe_frm_err_reasons[SPOE_APPCTX(appctx)->status_code]);
1869 			goto exit;
1870 
1871 		case  0: /* ignore */
1872 			goto next;
1873 
1874 		case  1: /* retry */
1875 			goto stop;
1876 
1877 		default:
1878 			SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1879 				    " - disconnected by peer (%d): %.*s\n",
1880 				    (int)now.tv_sec, (int)now.tv_usec,
1881 				    ((struct spoe_agent *)SPOE_APPCTX(appctx)->agent)->id,
1882 				    __FUNCTION__, appctx, SPOE_APPCTX(appctx)->status_code,
1883 				    SPOE_APPCTX(appctx)->rlen, SPOE_APPCTX(appctx)->reason);
1884 			goto exit;
1885 	}
1886 
1887   next:
1888 	/* Do not forget to remove processed frame from the output buffer */
1889 	if (trash.data)
1890 		co_skip(si_oc(appctx->owner), trash.data);
1891 
1892 	return 0;
1893   stop:
1894 	return 1;
1895   exit:
1896 	appctx->st0 = SPOE_APPCTX_ST_EXIT;
1897 	return 0;
1898 }
1899 
1900 /* I/O Handler processing messages exchanged with the agent */
1901 static void
spoe_handle_appctx(struct appctx * appctx)1902 spoe_handle_appctx(struct appctx *appctx)
1903 {
1904 	struct stream_interface *si = appctx->owner;
1905 	struct spoe_agent       *agent;
1906 
1907 	if (SPOE_APPCTX(appctx) == NULL)
1908 		return;
1909 
1910 	SPOE_APPCTX(appctx)->status_code = SPOE_FRM_ERR_NONE;
1911 	agent = SPOE_APPCTX(appctx)->agent;
1912 
1913   switchstate:
1914 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: appctx=%p"
1915 		    " - appctx-state=%s\n",
1916 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
1917 		    __FUNCTION__, appctx, spoe_appctx_state_str[appctx->st0]);
1918 
1919 	switch (appctx->st0) {
1920 		case SPOE_APPCTX_ST_CONNECT:
1921 			if (spoe_handle_connect_appctx(appctx))
1922 				goto out;
1923 			goto switchstate;
1924 
1925 		case SPOE_APPCTX_ST_CONNECTING:
1926 			if (spoe_handle_connecting_appctx(appctx))
1927 				goto out;
1928 			goto switchstate;
1929 
1930 		case SPOE_APPCTX_ST_IDLE:
1931 			_HA_ATOMIC_SUB(&agent->counters.idles, 1);
1932 			eb32_delete(&SPOE_APPCTX(appctx)->node);
1933 			if (stopping &&
1934 			    LIST_ISEMPTY(&agent->rt[tid].sending_queue) &&
1935 			    LIST_ISEMPTY(&SPOE_APPCTX(appctx)->waiting_queue)) {
1936 				SPOE_APPCTX(appctx)->task->expire =
1937 					tick_add_ifset(now_ms, agent->timeout.idle);
1938 				appctx->st0 = SPOE_APPCTX_ST_DISCONNECT;
1939 				goto switchstate;
1940 			}
1941 			appctx->st0 = SPOE_APPCTX_ST_PROCESSING;
1942 			/* fall through */
1943 
1944 		case SPOE_APPCTX_ST_PROCESSING:
1945 		case SPOE_APPCTX_ST_SENDING_FRAG_NOTIFY:
1946 		case SPOE_APPCTX_ST_WAITING_SYNC_ACK:
1947 			if (spoe_handle_processing_appctx(appctx))
1948 				goto out;
1949 			goto switchstate;
1950 
1951 		case SPOE_APPCTX_ST_DISCONNECT:
1952 			if (spoe_handle_disconnect_appctx(appctx))
1953 				goto out;
1954 			goto switchstate;
1955 
1956 		case SPOE_APPCTX_ST_DISCONNECTING:
1957 			if (spoe_handle_disconnecting_appctx(appctx))
1958 				goto out;
1959 			goto switchstate;
1960 
1961 		case SPOE_APPCTX_ST_EXIT:
1962 			appctx->st0 = SPOE_APPCTX_ST_END;
1963 			SPOE_APPCTX(appctx)->task->expire = TICK_ETERNITY;
1964 
1965 			si_shutw(si);
1966 			si_shutr(si);
1967 			si_ic(si)->flags |= CF_READ_NULL;
1968 			/* fall through */
1969 
1970 		case SPOE_APPCTX_ST_END:
1971 			return;
1972 	}
1973   out:
1974 	if (stopping)
1975 		spoe_wakeup_appctx(appctx);
1976 
1977 	if (SPOE_APPCTX(appctx)->task->expire != TICK_ETERNITY)
1978 		task_queue(SPOE_APPCTX(appctx)->task);
1979 }
1980 
1981 struct applet spoe_applet = {
1982 	.obj_type = OBJ_TYPE_APPLET,
1983 	.name = "<SPOE>", /* used for logging */
1984 	.fct = spoe_handle_appctx,
1985 	.release = spoe_release_appctx,
1986 };
1987 
1988 /* Create a SPOE applet. On success, the created applet is returned, else
1989  * NULL. */
1990 static struct appctx *
spoe_create_appctx(struct spoe_config * conf)1991 spoe_create_appctx(struct spoe_config *conf)
1992 {
1993 	struct appctx      *appctx;
1994 	struct session     *sess;
1995 	struct stream      *strm;
1996 
1997 	if ((appctx = appctx_new(&spoe_applet, tid_bit)) == NULL)
1998 		goto out_error;
1999 
2000 	appctx->ctx.spoe.ptr = pool_alloc_dirty(pool_head_spoe_appctx);
2001 	if (SPOE_APPCTX(appctx) == NULL)
2002 		goto out_free_appctx;
2003 	memset(appctx->ctx.spoe.ptr, 0, pool_head_spoe_appctx->size);
2004 
2005 	appctx->st0 = SPOE_APPCTX_ST_CONNECT;
2006 	if ((SPOE_APPCTX(appctx)->task = task_new(tid_bit)) == NULL)
2007 		goto out_free_spoe_appctx;
2008 
2009 	SPOE_APPCTX(appctx)->owner           = appctx;
2010 	SPOE_APPCTX(appctx)->task->process   = spoe_process_appctx;
2011 	SPOE_APPCTX(appctx)->task->context   = appctx;
2012 	SPOE_APPCTX(appctx)->agent           = conf->agent;
2013 	SPOE_APPCTX(appctx)->version         = 0;
2014 	SPOE_APPCTX(appctx)->max_frame_size  = conf->agent->max_frame_size;
2015 	SPOE_APPCTX(appctx)->flags           = 0;
2016 	SPOE_APPCTX(appctx)->status_code     = SPOE_FRM_ERR_NONE;
2017 	SPOE_APPCTX(appctx)->buffer          = BUF_NULL;
2018 	SPOE_APPCTX(appctx)->cur_fpa         = 0;
2019 
2020 	LIST_INIT(&SPOE_APPCTX(appctx)->buffer_wait.list);
2021 	SPOE_APPCTX(appctx)->buffer_wait.target = appctx;
2022 	SPOE_APPCTX(appctx)->buffer_wait.wakeup_cb = (int (*)(void *))spoe_wakeup_appctx;
2023 
2024 	LIST_INIT(&SPOE_APPCTX(appctx)->list);
2025 	LIST_INIT(&SPOE_APPCTX(appctx)->waiting_queue);
2026 
2027 	sess = session_new(&conf->agent_fe, NULL, &appctx->obj_type);
2028 	if (!sess)
2029 		goto out_free_spoe;
2030 
2031 	if ((strm = stream_new(sess, &appctx->obj_type)) == NULL)
2032 		goto out_free_sess;
2033 
2034 	stream_set_backend(strm, conf->agent->b.be);
2035 
2036 	/* applet is waiting for data */
2037 	si_cant_get(&strm->si[0]);
2038 	appctx_wakeup(appctx);
2039 
2040 	strm->do_log = NULL;
2041 	strm->res.flags |= CF_READ_DONTWAIT;
2042 
2043 	HA_SPIN_LOCK(SPOE_APPLET_LOCK, &conf->agent->rt[tid].lock);
2044 	LIST_ADDQ(&conf->agent->rt[tid].applets, &SPOE_APPCTX(appctx)->list);
2045 	HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &conf->agent->rt[tid].lock);
2046 	_HA_ATOMIC_ADD(&conf->agent->counters.applets, 1);
2047 
2048 	task_wakeup(SPOE_APPCTX(appctx)->task, TASK_WOKEN_INIT);
2049 	task_wakeup(strm->task, TASK_WOKEN_INIT);
2050 	return appctx;
2051 
2052 	/* Error unrolling */
2053  out_free_sess:
2054 	session_free(sess);
2055  out_free_spoe:
2056 	task_destroy(SPOE_APPCTX(appctx)->task);
2057  out_free_spoe_appctx:
2058 	pool_free(pool_head_spoe_appctx, SPOE_APPCTX(appctx));
2059  out_free_appctx:
2060 	appctx_free(appctx);
2061  out_error:
2062 	return NULL;
2063 }
2064 
2065 static int
spoe_queue_context(struct spoe_context * ctx)2066 spoe_queue_context(struct spoe_context *ctx)
2067 {
2068 	struct spoe_config *conf = FLT_CONF(ctx->filter);
2069 	struct spoe_agent  *agent = conf->agent;
2070 	struct appctx      *appctx;
2071 	struct spoe_appctx *spoe_appctx;
2072 
2073 	/* Check if we need to create a new SPOE applet or not. */
2074 	if (!eb_is_empty(&agent->rt[tid].idle_applets) &&
2075 	    (agent->rt[tid].processing == 1 || agent->rt[tid].processing < read_freq_ctr(&agent->rt[tid].processing_per_sec)))
2076 		goto end;
2077 
2078 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2079 		    " - try to create new SPOE appctx\n",
2080 		    (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__,
2081 		    ctx->strm);
2082 
2083 	/* Do not try to create a new applet if there is no server up for the
2084 	 * agent's backend. */
2085 	if (!agent->b.be->srv_act && !agent->b.be->srv_bck) {
2086 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2087 			    " - cannot create SPOE appctx: no server up\n",
2088 			    (int)now.tv_sec, (int)now.tv_usec, agent->id,
2089 			    __FUNCTION__, ctx->strm);
2090 		goto end;
2091 	}
2092 
2093 	/* Do not try to create a new applet if we have reached the maximum of
2094 	 * connection per seconds */
2095 	if (agent->cps_max > 0) {
2096 		if (!freq_ctr_remain(&agent->rt[tid].conn_per_sec, agent->cps_max, 0)) {
2097 			SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2098 				    " - cannot create SPOE appctx: max CPS reached\n",
2099 				    (int)now.tv_sec, (int)now.tv_usec, agent->id,
2100 				    __FUNCTION__, ctx->strm);
2101 			goto end;
2102 		}
2103 	}
2104 
2105 	appctx = spoe_create_appctx(conf);
2106 	if (appctx == NULL) {
2107 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2108 			    " - failed to create SPOE appctx\n",
2109 			    (int)now.tv_sec, (int)now.tv_usec, agent->id,
2110 			    __FUNCTION__, ctx->strm);
2111 		send_log(&conf->agent_fe, LOG_EMERG,
2112 			 "SPOE: [%s] failed to create SPOE applet\n",
2113 			 agent->id);
2114 
2115 		goto end;
2116 	}
2117 
2118 	/* Increase the per-process number of cumulated connections */
2119 	if (agent->cps_max > 0)
2120 		update_freq_ctr(&agent->rt[tid].conn_per_sec, 1);
2121 
2122   end:
2123 	/* The only reason to return an error is when there is no applet */
2124 	if (LIST_ISEMPTY(&agent->rt[tid].applets)) {
2125 		ctx->status_code = SPOE_CTX_ERR_RES;
2126 		return -1;
2127 	}
2128 
2129 	/* Add the SPOE context in the sending queue if the stream has no applet
2130 	 * already assigned and wakeup all idle applets. Otherwise, don't queue
2131 	 * it. */
2132 	_HA_ATOMIC_ADD(&agent->counters.nb_sending, 1);
2133 	spoe_update_stat_time(&ctx->stats.tv_request, &ctx->stats.t_request);
2134 	ctx->stats.tv_queue = now;
2135 	if (ctx->spoe_appctx)
2136 		return 1;
2137 	LIST_ADDQ(&agent->rt[tid].sending_queue, &ctx->list);
2138 
2139 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2140 		    " - Add stream in sending queue"
2141 		    " - applets=%u - idles=%u - processing=%u\n",
2142 		    (int)now.tv_sec, (int)now.tv_usec, agent->id, __FUNCTION__,
2143 		    ctx->strm, agent->counters.applets, agent->counters.idles,
2144 		    agent->rt[tid].processing);
2145 
2146 	/* Finally try to wakeup an IDLE applet. */
2147 	if (!eb_is_empty(&agent->rt[tid].idle_applets)) {
2148 		struct eb32_node *node;
2149 
2150 		node = eb32_first(&agent->rt[tid].idle_applets);
2151 		spoe_appctx = eb32_entry(node, struct spoe_appctx, node);
2152 		if (node && spoe_appctx) {
2153 			eb32_delete(&spoe_appctx->node);
2154 			spoe_appctx->node.key++;
2155 			eb32_insert(&agent->rt[tid].idle_applets, &spoe_appctx->node);
2156 			spoe_wakeup_appctx(spoe_appctx->owner);
2157 		}
2158 	}
2159 	return 1;
2160 }
2161 
2162 /***************************************************************************
2163  * Functions that encode SPOE messages
2164  **************************************************************************/
2165 /* Encode a SPOE message. Info in <ctx->frag_ctx>, if any, are used to handle
2166  * fragmented_content. If the next message can be processed, it returns 0. If
2167  * the message is too big, it returns -1.*/
2168 static int
spoe_encode_message(struct stream * s,struct spoe_context * ctx,struct spoe_message * msg,int dir,char ** buf,char * end)2169 spoe_encode_message(struct stream *s, struct spoe_context *ctx,
2170 		    struct spoe_message *msg, int dir,
2171 		    char **buf, char *end)
2172 {
2173 	struct sample   *smp;
2174 	struct spoe_arg *arg;
2175 	int ret;
2176 
2177 	if (msg->cond) {
2178 		ret = acl_exec_cond(msg->cond, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2179 		ret = acl_pass(ret);
2180 		if (msg->cond->pol == ACL_COND_UNLESS)
2181 			ret = !ret;
2182 
2183 		/* the rule does not match */
2184 		if (!ret)
2185 			goto next;
2186 	}
2187 
2188 		/* Resume encoding of a SPOE argument */
2189 	if (ctx->frag_ctx.curarg != NULL) {
2190 		arg = ctx->frag_ctx.curarg;
2191 		goto encode_argument;
2192 	}
2193 
2194 	if (ctx->frag_ctx.curoff != UINT_MAX)
2195 		goto encode_msg_payload;
2196 
2197 	/* Check if there is enough space for the message name and the
2198 	 * number of arguments. It implies <msg->id_len> is encoded on 2
2199 	 * bytes, at most (< 2288). */
2200 	if (*buf + 2 + msg->id_len + 1 > end)
2201 		goto too_big;
2202 
2203 	/* Encode the message name */
2204 	if (spoe_encode_buffer(msg->id, msg->id_len, buf, end) == -1)
2205 		goto too_big;
2206 
2207 	/* Set the number of arguments for this message */
2208 	**buf = msg->nargs;
2209 	(*buf)++;
2210 
2211 	ctx->frag_ctx.curoff = 0;
2212   encode_msg_payload:
2213 
2214 	/* Loop on arguments */
2215 	list_for_each_entry(arg, &msg->args, list) {
2216 		ctx->frag_ctx.curarg = arg;
2217 		ctx->frag_ctx.curoff = UINT_MAX;
2218 		ctx->frag_ctx.curlen = 0;
2219 
2220 	  encode_argument:
2221 		if (ctx->frag_ctx.curoff != UINT_MAX)
2222 			goto encode_arg_value;
2223 
2224 		/* Encode the argument name as a string. It can by NULL */
2225 		if (spoe_encode_buffer(arg->name, arg->name_len, buf, end) == -1)
2226 			goto too_big;
2227 
2228 		ctx->frag_ctx.curoff = 0;
2229 	  encode_arg_value:
2230 
2231 		/* Fetch the argument value */
2232 		smp = sample_process(s->be, s->sess, s, dir|SMP_OPT_FINAL, arg->expr, NULL);
2233 		if (smp) {
2234 			smp->ctx.a[0] = &ctx->frag_ctx.curlen;
2235 			smp->ctx.a[1] = &ctx->frag_ctx.curoff;
2236 		}
2237 		ret = spoe_encode_data(smp, buf, end);
2238 		if (ret == -1 || ctx->frag_ctx.curoff)
2239 			goto too_big;
2240 	}
2241 
2242   next:
2243 	return 0;
2244 
2245   too_big:
2246 	return -1;
2247 }
2248 
2249 /* Encode list of SPOE messages. Info in <ctx->frag_ctx>, if any, are used to
2250  * handle fragmented content. On success it returns 1. If an error occurred, -1
2251  * is returned. If nothing has been encoded, it returns 0 (this is only possible
2252  * for unfragmented payload). */
2253 static int
spoe_encode_messages(struct stream * s,struct spoe_context * ctx,struct list * messages,int dir,int type)2254 spoe_encode_messages(struct stream *s, struct spoe_context *ctx,
2255 		     struct list *messages, int dir, int type)
2256 {
2257 	struct spoe_config  *conf = FLT_CONF(ctx->filter);
2258 	struct spoe_agent   *agent = conf->agent;
2259 	struct spoe_message *msg;
2260 	char   *p, *end;
2261 
2262 	p   = b_head(&ctx->buffer);
2263 	end =  p + agent->rt[tid].frame_size - FRAME_HDR_SIZE;
2264 
2265 	if (type == SPOE_MSGS_BY_EVENT) { /* Loop on messages by event */
2266 		/* Resume encoding of a SPOE message */
2267 		if (ctx->frag_ctx.curmsg != NULL) {
2268 			msg = ctx->frag_ctx.curmsg;
2269 			goto encode_evt_message;
2270 		}
2271 
2272 		list_for_each_entry(msg, messages, by_evt) {
2273 			ctx->frag_ctx.curmsg = msg;
2274 			ctx->frag_ctx.curarg = NULL;
2275 			ctx->frag_ctx.curoff = UINT_MAX;
2276 
2277 		encode_evt_message:
2278 			if (spoe_encode_message(s, ctx, msg, dir, &p, end) == -1)
2279 				goto too_big;
2280 		}
2281 	}
2282 	else if (type == SPOE_MSGS_BY_GROUP) { /* Loop on messages by group */
2283 		/* Resume encoding of a SPOE message */
2284 		if (ctx->frag_ctx.curmsg != NULL) {
2285 			msg = ctx->frag_ctx.curmsg;
2286 			goto encode_grp_message;
2287 		}
2288 
2289 		list_for_each_entry(msg, messages, by_grp) {
2290 			ctx->frag_ctx.curmsg = msg;
2291 			ctx->frag_ctx.curarg = NULL;
2292 			ctx->frag_ctx.curoff = UINT_MAX;
2293 
2294 		encode_grp_message:
2295 			if (spoe_encode_message(s, ctx, msg, dir, &p, end) == -1)
2296 				goto too_big;
2297 		}
2298 	}
2299 	else
2300 		goto skip;
2301 
2302 
2303 	/* nothing has been encoded for an unfragmented payload */
2304 	if (!(ctx->flags & SPOE_CTX_FL_FRAGMENTED) && p == b_head(&ctx->buffer))
2305 		goto skip;
2306 
2307 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2308 		    " - encode %s messages - spoe_appctx=%p"
2309 		    "- max_size=%u - encoded=%ld\n",
2310 		    (int)now.tv_sec, (int)now.tv_usec,
2311 		    agent->id, __FUNCTION__, s,
2312 		    ((ctx->flags & SPOE_CTX_FL_FRAGMENTED) ? "last fragment of" : "unfragmented"),
2313 		    ctx->spoe_appctx, (agent->rt[tid].frame_size - FRAME_HDR_SIZE),
2314 		    p - b_head(&ctx->buffer));
2315 
2316 	b_set_data(&ctx->buffer, p - b_head(&ctx->buffer));
2317 	ctx->frag_ctx.curmsg = NULL;
2318 	ctx->frag_ctx.curarg = NULL;
2319 	ctx->frag_ctx.curoff = 0;
2320 	ctx->frag_ctx.flags  = SPOE_FRM_FL_FIN;
2321 
2322 	return 1;
2323 
2324   too_big:
2325 	/* Return an error if fragmentation is unsupported or if nothing has
2326 	 * been encoded because its too big and not splittable. */
2327 	if (!(agent->flags & SPOE_FL_SND_FRAGMENTATION) || p == b_head(&ctx->buffer)) {
2328 		ctx->status_code = SPOE_CTX_ERR_TOO_BIG;
2329 		return -1;
2330 	}
2331 
2332 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2333 		    " - encode fragmented messages - spoe_appctx=%p"
2334 		    " - curmsg=%p - curarg=%p - curoff=%u"
2335 		    " - max_size=%u - encoded=%ld\n",
2336 		    (int)now.tv_sec, (int)now.tv_usec,
2337 		    agent->id, __FUNCTION__, s, ctx->spoe_appctx,
2338 		    ctx->frag_ctx.curmsg, ctx->frag_ctx.curarg, ctx->frag_ctx.curoff,
2339 		    (agent->rt[tid].frame_size - FRAME_HDR_SIZE), p - b_head(&ctx->buffer));
2340 
2341 	b_set_data(&ctx->buffer, p - b_head(&ctx->buffer));
2342 	ctx->flags |= SPOE_CTX_FL_FRAGMENTED;
2343 	ctx->frag_ctx.flags &= ~SPOE_FRM_FL_FIN;
2344 	return 1;
2345 
2346   skip:
2347 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2348 		    " - skip the frame because nothing has been encoded\n",
2349 		    (int)now.tv_sec, (int)now.tv_usec,
2350 		    agent->id, __FUNCTION__, s);
2351 	return 0;
2352 }
2353 
2354 
2355 /***************************************************************************
2356  * Functions that handle SPOE actions
2357  **************************************************************************/
2358 /* Helper function to set a variable */
2359 static void
spoe_set_var(struct spoe_context * ctx,char * scope,char * name,int len,struct sample * smp)2360 spoe_set_var(struct spoe_context *ctx, char *scope, char *name, int len,
2361 	     struct sample *smp)
2362 {
2363 	struct spoe_config *conf = FLT_CONF(ctx->filter);
2364 	struct spoe_agent  *agent = conf->agent;
2365 	char                varname[64];
2366 
2367 	memset(varname, 0, sizeof(varname));
2368 	len = snprintf(varname, sizeof(varname), "%s.%s.%.*s",
2369 		       scope, agent->var_pfx, len, name);
2370 	if (agent->flags & SPOE_FL_FORCE_SET_VAR)
2371 		vars_set_by_name(varname, len, smp);
2372 	else
2373 		vars_set_by_name_ifexist(varname, len, smp);
2374 }
2375 
2376 /* Helper function to unset a variable */
2377 static void
spoe_unset_var(struct spoe_context * ctx,char * scope,char * name,int len,struct sample * smp)2378 spoe_unset_var(struct spoe_context *ctx, char *scope, char *name, int len,
2379 	       struct sample *smp)
2380 {
2381 	struct spoe_config *conf = FLT_CONF(ctx->filter);
2382 	struct spoe_agent  *agent = conf->agent;
2383 	char                varname[64];
2384 
2385 	memset(varname, 0, sizeof(varname));
2386 	len = snprintf(varname, sizeof(varname), "%s.%s.%.*s",
2387 		       scope, agent->var_pfx, len, name);
2388 	vars_unset_by_name_ifexist(varname, len, smp);
2389 }
2390 
2391 
2392 static inline int
spoe_decode_action_set_var(struct stream * s,struct spoe_context * ctx,char ** buf,char * end,int dir)2393 spoe_decode_action_set_var(struct stream *s, struct spoe_context *ctx,
2394 			   char **buf, char *end, int dir)
2395 {
2396 	char         *str, *scope, *p = *buf;
2397 	struct sample smp;
2398 	uint64_t      sz;
2399 	int           ret;
2400 
2401 	if (p + 2 >= end)
2402 		goto skip;
2403 
2404 	/* SET-VAR requires 3 arguments */
2405 	if (*p++ != 3)
2406 		goto skip;
2407 
2408 	switch (*p++) {
2409 		case SPOE_SCOPE_PROC: scope = "proc"; break;
2410 		case SPOE_SCOPE_SESS: scope = "sess"; break;
2411 		case SPOE_SCOPE_TXN : scope = "txn";  break;
2412 		case SPOE_SCOPE_REQ : scope = "req";  break;
2413 		case SPOE_SCOPE_RES : scope = "res";  break;
2414 		default: goto skip;
2415 	}
2416 
2417 	if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
2418 		goto skip;
2419 	memset(&smp, 0, sizeof(smp));
2420 	smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2421 
2422 	if (spoe_decode_data(&p, end, &smp) == -1)
2423 		goto skip;
2424 
2425 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2426 		    " - set-var '%s.%s.%.*s'\n",
2427 		    (int)now.tv_sec, (int)now.tv_usec,
2428 		    ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->id,
2429 		    __FUNCTION__, s, scope,
2430 		    ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->var_pfx,
2431 		    (int)sz, str);
2432 
2433 	if (smp.data.type == SMP_T_ANY)
2434 		spoe_unset_var(ctx, scope, str, sz, &smp);
2435 	else
2436 		spoe_set_var(ctx, scope, str, sz, &smp);
2437 
2438 	ret  = (p - *buf);
2439 	*buf = p;
2440 	return ret;
2441   skip:
2442 	return 0;
2443 }
2444 
2445 static inline int
spoe_decode_action_unset_var(struct stream * s,struct spoe_context * ctx,char ** buf,char * end,int dir)2446 spoe_decode_action_unset_var(struct stream *s, struct spoe_context *ctx,
2447 			     char **buf, char *end, int dir)
2448 {
2449 	char         *str, *scope, *p = *buf;
2450 	struct sample smp;
2451 	uint64_t      sz;
2452 	int           ret;
2453 
2454 	if (p + 2 >= end)
2455 		goto skip;
2456 
2457 	/* UNSET-VAR requires 2 arguments */
2458 	if (*p++ != 2)
2459 		goto skip;
2460 
2461 	switch (*p++) {
2462 		case SPOE_SCOPE_PROC: scope = "proc"; break;
2463 		case SPOE_SCOPE_SESS: scope = "sess"; break;
2464 		case SPOE_SCOPE_TXN : scope = "txn";  break;
2465 		case SPOE_SCOPE_REQ : scope = "req";  break;
2466 		case SPOE_SCOPE_RES : scope = "res";  break;
2467 		default: goto skip;
2468 	}
2469 
2470 	if (spoe_decode_buffer(&p, end, &str, &sz) == -1)
2471 		goto skip;
2472 	memset(&smp, 0, sizeof(smp));
2473 	smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2474 
2475 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2476 		    " - unset-var '%s.%s.%.*s'\n",
2477 		    (int)now.tv_sec, (int)now.tv_usec,
2478 		    ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->id,
2479 		    __FUNCTION__, s, scope,
2480 		    ((struct spoe_config *)FLT_CONF(ctx->filter))->agent->var_pfx,
2481 		    (int)sz, str);
2482 
2483 	spoe_unset_var(ctx, scope, str, sz, &smp);
2484 
2485 	ret  = (p - *buf);
2486 	*buf = p;
2487 	return ret;
2488   skip:
2489 	return 0;
2490 }
2491 
2492 /* Process SPOE actions for a specific event. It returns 1 on success. If an
2493  * error occurred, 0 is returned. */
2494 static int
spoe_process_actions(struct stream * s,struct spoe_context * ctx,int dir)2495 spoe_process_actions(struct stream *s, struct spoe_context *ctx, int dir)
2496 {
2497 	char *p, *end;
2498 	int   ret;
2499 
2500 	p   = b_head(&ctx->buffer);
2501 	end = p + b_data(&ctx->buffer);
2502 
2503 	while (p < end)  {
2504 		enum spoe_action_type type;
2505 
2506 		type = *p++;
2507 		switch (type) {
2508 			case SPOE_ACT_T_SET_VAR:
2509 				ret = spoe_decode_action_set_var(s, ctx, &p, end, dir);
2510 				if (!ret)
2511 					goto skip;
2512 				break;
2513 
2514 			case SPOE_ACT_T_UNSET_VAR:
2515 				ret = spoe_decode_action_unset_var(s, ctx, &p, end, dir);
2516 				if (!ret)
2517 					goto skip;
2518 				break;
2519 
2520 			default:
2521 				goto skip;
2522 		}
2523 	}
2524 
2525 	return 1;
2526   skip:
2527 	return 0;
2528 }
2529 
2530 /***************************************************************************
2531  * Functions that process SPOE events
2532  **************************************************************************/
2533 static void
spoe_update_stats(struct stream * s,struct spoe_agent * agent,struct spoe_context * ctx,int dir)2534 spoe_update_stats(struct stream *s, struct spoe_agent *agent,
2535 		  struct spoe_context *ctx, int dir)
2536 {
2537 	if (!tv_iszero(&ctx->stats.tv_start)) {
2538 		spoe_update_stat_time(&ctx->stats.tv_start, &ctx->stats.t_process);
2539 		ctx->stats.t_total  += ctx->stats.t_process;
2540 		tv_zero(&ctx->stats.tv_request);
2541 		tv_zero(&ctx->stats.tv_queue);
2542 		tv_zero(&ctx->stats.tv_wait);
2543 		tv_zero(&ctx->stats.tv_response);
2544 	}
2545 
2546 	if (agent->var_t_process) {
2547 		struct sample smp;
2548 
2549 		memset(&smp, 0, sizeof(smp));
2550 		smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2551 		smp.data.u.sint = ctx->stats.t_process;
2552 		smp.data.type   = SMP_T_SINT;
2553 
2554 		spoe_set_var(ctx, "txn", agent->var_t_process,
2555 			     strlen(agent->var_t_process), &smp);
2556 	}
2557 
2558 	if (agent->var_t_total) {
2559 		struct sample smp;
2560 
2561 		memset(&smp, 0, sizeof(smp));
2562 		smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2563 		smp.data.u.sint = ctx->stats.t_total;
2564 		smp.data.type   = SMP_T_SINT;
2565 
2566 		spoe_set_var(ctx, "txn", agent->var_t_total,
2567 			     strlen(agent->var_t_total), &smp);
2568 	}
2569 }
2570 
2571 static void
spoe_handle_processing_error(struct stream * s,struct spoe_agent * agent,struct spoe_context * ctx,int dir)2572 spoe_handle_processing_error(struct stream *s, struct spoe_agent *agent,
2573 			     struct spoe_context *ctx, int dir)
2574 {
2575 	if (agent->eps_max > 0)
2576 		update_freq_ctr(&agent->rt[tid].err_per_sec, 1);
2577 
2578 	if (agent->var_on_error) {
2579 		struct sample smp;
2580 
2581 		memset(&smp, 0, sizeof(smp));
2582 		smp_set_owner(&smp, s->be, s->sess, s, dir|SMP_OPT_FINAL);
2583 		smp.data.u.sint = ctx->status_code;
2584 		smp.data.type   = SMP_T_BOOL;
2585 
2586 		spoe_set_var(ctx, "txn", agent->var_on_error,
2587 			     strlen(agent->var_on_error), &smp);
2588 	}
2589 
2590 	ctx->state = ((agent->flags & SPOE_FL_CONT_ON_ERR)
2591 		      ? SPOE_CTX_ST_READY
2592 		      : SPOE_CTX_ST_NONE);
2593 }
2594 
2595 static inline int
spoe_start_processing(struct spoe_agent * agent,struct spoe_context * ctx,int dir)2596 spoe_start_processing(struct spoe_agent *agent, struct spoe_context *ctx, int dir)
2597 {
2598 	/* If a process is already started for this SPOE context, retry
2599 	 * later. */
2600 	if (ctx->flags & SPOE_CTX_FL_PROCESS)
2601 		return 0;
2602 
2603 	agent->rt[tid].processing++;
2604 	ctx->stats.tv_start   = now;
2605 	ctx->stats.tv_request = now;
2606 	ctx->stats.t_request  = -1;
2607 	ctx->stats.t_queue    = -1;
2608 	ctx->stats.t_waiting  = -1;
2609 	ctx->stats.t_response = -1;
2610 	ctx->stats.t_process  = -1;
2611 
2612 	ctx->status_code = 0;
2613 
2614 	/* Set the right flag to prevent request and response processing
2615 	 * in same time. */
2616 	ctx->flags |= ((dir == SMP_OPT_DIR_REQ)
2617 		       ? SPOE_CTX_FL_REQ_PROCESS
2618 		       : SPOE_CTX_FL_RSP_PROCESS);
2619 	return 1;
2620 }
2621 
2622 static inline void
spoe_stop_processing(struct spoe_agent * agent,struct spoe_context * ctx)2623 spoe_stop_processing(struct spoe_agent *agent, struct spoe_context *ctx)
2624 {
2625 	struct spoe_appctx *sa = ctx->spoe_appctx;
2626 
2627 	if (!(ctx->flags & SPOE_CTX_FL_PROCESS))
2628 		return;
2629 	_HA_ATOMIC_ADD(&agent->counters.nb_processed, 1);
2630 	if (sa) {
2631 		if (sa->frag_ctx.ctx == ctx) {
2632 			sa->frag_ctx.ctx = NULL;
2633 			spoe_wakeup_appctx(sa->owner);
2634 		}
2635 		else
2636 			sa->cur_fpa--;
2637 	}
2638 
2639 	/* Reset the flag to allow next processing */
2640 	agent->rt[tid].processing--;
2641 	ctx->flags &= ~(SPOE_CTX_FL_PROCESS|SPOE_CTX_FL_FRAGMENTED);
2642 
2643 	/* Reset processing timer */
2644 	ctx->process_exp = TICK_ETERNITY;
2645 
2646 	spoe_release_buffer(&ctx->buffer, &ctx->buffer_wait);
2647 
2648 	ctx->spoe_appctx          = NULL;
2649 	ctx->frag_ctx.curmsg      = NULL;
2650 	ctx->frag_ctx.curarg      = NULL;
2651 	ctx->frag_ctx.curoff      = 0;
2652 	ctx->frag_ctx.flags       = 0;
2653 
2654 	if (!LIST_ISEMPTY(&ctx->list)) {
2655 		if (ctx->state == SPOE_CTX_ST_SENDING_MSGS)
2656 			_HA_ATOMIC_SUB(&agent->counters.nb_sending, 1);
2657 		else
2658 			_HA_ATOMIC_SUB(&agent->counters.nb_waiting, 1);
2659 
2660 		LIST_DEL(&ctx->list);
2661 		LIST_INIT(&ctx->list);
2662 	}
2663 }
2664 
2665 /* Process a list of SPOE messages. First, this functions will process messages
2666  *  and send them to an agent in a NOTIFY frame. Then, it will wait a ACK frame
2667  *  to process corresponding actions. During all the processing, it returns 0
2668  *  and it returns 1 when the processing is finished. If an error occurred, -1
2669  *  is returned. */
2670 static int
spoe_process_messages(struct stream * s,struct spoe_context * ctx,struct list * messages,int dir,int type)2671 spoe_process_messages(struct stream *s, struct spoe_context *ctx,
2672 		      struct list *messages, int dir, int type)
2673 {
2674 	struct spoe_config *conf = FLT_CONF(ctx->filter);
2675 	struct spoe_agent  *agent = conf->agent;
2676 	int                 ret = 1;
2677 
2678 	if (ctx->state == SPOE_CTX_ST_ERROR)
2679 		goto end;
2680 
2681 	if (tick_is_expired(ctx->process_exp, now_ms) && ctx->state != SPOE_CTX_ST_DONE) {
2682 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2683 			    " - failed to process messages: timeout\n",
2684 			    (int)now.tv_sec, (int)now.tv_usec,
2685 			    agent->id, __FUNCTION__, s);
2686 		ctx->status_code = SPOE_CTX_ERR_TOUT;
2687 		goto end;
2688 	}
2689 
2690 	if (ctx->state == SPOE_CTX_ST_READY) {
2691 		if (agent->eps_max > 0) {
2692 			if (!freq_ctr_remain(&agent->rt[tid].err_per_sec, agent->eps_max, 0)) {
2693 				SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2694 					    " - skip processing of messages: max EPS reached\n",
2695 					    (int)now.tv_sec, (int)now.tv_usec,
2696 					    agent->id, __FUNCTION__, s);
2697 				goto skip;
2698 			}
2699 		}
2700 
2701 		if (!tick_isset(ctx->process_exp)) {
2702 			ctx->process_exp = tick_add_ifset(now_ms, agent->timeout.processing);
2703 			s->task->expire  = tick_first((tick_is_expired(s->task->expire, now_ms) ? 0 : s->task->expire),
2704 						      ctx->process_exp);
2705 		}
2706 		ret = spoe_start_processing(agent, ctx, dir);
2707 		if (!ret)
2708 			goto out;
2709 
2710 		ctx->state = SPOE_CTX_ST_ENCODING_MSGS;
2711 		/* fall through */
2712 	}
2713 
2714 	if (ctx->state == SPOE_CTX_ST_ENCODING_MSGS) {
2715 		if (tv_iszero(&ctx->stats.tv_request))
2716 			ctx->stats.tv_request = now;
2717 		if (!spoe_acquire_buffer(&ctx->buffer, &ctx->buffer_wait))
2718 			goto out;
2719 		ret = spoe_encode_messages(s, ctx, messages, dir, type);
2720 		if (ret < 0)
2721 			goto end;
2722 		if (!ret)
2723 			goto skip;
2724 		if (spoe_queue_context(ctx) < 0)
2725 			goto end;
2726 		ctx->state = SPOE_CTX_ST_SENDING_MSGS;
2727 	}
2728 
2729 	if (ctx->state == SPOE_CTX_ST_SENDING_MSGS) {
2730 		if (ctx->spoe_appctx)
2731 			spoe_wakeup_appctx(ctx->spoe_appctx->owner);
2732 		ret = 0;
2733 		goto out;
2734 	}
2735 
2736 	if (ctx->state == SPOE_CTX_ST_WAITING_ACK) {
2737 		ret = 0;
2738 		goto out;
2739 	}
2740 
2741 	if (ctx->state == SPOE_CTX_ST_DONE) {
2742 		spoe_process_actions(s, ctx, dir);
2743 		ret = 1;
2744 		ctx->frame_id++;
2745 		ctx->state = SPOE_CTX_ST_READY;
2746 		spoe_update_stat_time(&ctx->stats.tv_response, &ctx->stats.t_response);
2747 		goto end;
2748 	}
2749 
2750   out:
2751 	return ret;
2752 
2753   skip:
2754 	tv_zero(&ctx->stats.tv_start);
2755 	ctx->state = SPOE_CTX_ST_READY;
2756 	spoe_stop_processing(agent, ctx);
2757 	return 1;
2758 
2759   end:
2760 	spoe_update_stats(s, agent, ctx, dir);
2761 	spoe_stop_processing(agent, ctx);
2762 	if (ctx->status_code) {
2763 		_HA_ATOMIC_ADD(&agent->counters.nb_errors, 1);
2764 		spoe_handle_processing_error(s, agent, ctx, dir);
2765 		ret = 1;
2766 	}
2767 	return ret;
2768 }
2769 
2770 /* Process a SPOE group, ie the list of messages attached to the group <grp>.
2771  * See spoe_process_message for details. */
2772 static int
spoe_process_group(struct stream * s,struct spoe_context * ctx,struct spoe_group * group,int dir)2773 spoe_process_group(struct stream *s, struct spoe_context *ctx,
2774 		   struct spoe_group *group, int dir)
2775 {
2776 	struct spoe_config *conf = FLT_CONF(ctx->filter);
2777 	struct spoe_agent  *agent = conf->agent;
2778 	int ret;
2779 
2780 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2781 		    " - ctx-state=%s - Process messages for group=%s\n",
2782 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
2783 		    __FUNCTION__, s, spoe_ctx_state_str[ctx->state],
2784 		    group->id);
2785 
2786 	if (LIST_ISEMPTY(&group->messages))
2787 		return 1;
2788 
2789 	ret = spoe_process_messages(s, ctx, &group->messages, dir, SPOE_MSGS_BY_GROUP);
2790 	if (ret && ctx->stats.t_process != -1) {
2791 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2792 			    " - <GROUP:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u/%u\n",
2793 			    (int)now.tv_sec, (int)now.tv_usec, agent->id,
2794 			    __FUNCTION__, s, group->id, s->uniq_id, ctx->status_code,
2795 			    ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2796 			    ctx->stats.t_response, ctx->stats.t_process,
2797 			    agent->counters.idles, agent->counters.applets,
2798 			    agent->counters.nb_sending, agent->counters.nb_waiting,
2799 			    agent->counters.nb_errors, agent->counters.nb_processed,
2800 			    agent->rt[tid].processing, read_freq_ctr(&agent->rt[tid].processing_per_sec));
2801 		if (ctx->status_code || !(conf->agent_fe.options2 & PR_O2_NOLOGNORM))
2802 			send_log(&conf->agent_fe, (!ctx->status_code ? LOG_NOTICE : LOG_WARNING),
2803 				 "SPOE: [%s] <GROUP:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu\n",
2804 				 agent->id, group->id, s->uniq_id, ctx->status_code,
2805 				 ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2806 				 ctx->stats.t_response, ctx->stats.t_process,
2807 				 agent->counters.idles, agent->counters.applets,
2808 				 agent->counters.nb_sending, agent->counters.nb_waiting,
2809 				 agent->counters.nb_errors, agent->counters.nb_processed);
2810 	}
2811 	return ret;
2812 }
2813 
2814 /* Process a SPOE event, ie the list of messages attached to the event <ev>.
2815  * See spoe_process_message for details. */
2816 static int
spoe_process_event(struct stream * s,struct spoe_context * ctx,enum spoe_event ev)2817 spoe_process_event(struct stream *s, struct spoe_context *ctx,
2818 		   enum spoe_event ev)
2819 {
2820 	struct spoe_config *conf = FLT_CONF(ctx->filter);
2821 	struct spoe_agent  *agent = conf->agent;
2822 	int dir, ret;
2823 
2824 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2825 		    " - ctx-state=%s - Process messages for event=%s\n",
2826 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
2827 		    __FUNCTION__, s, spoe_ctx_state_str[ctx->state],
2828 		    spoe_event_str[ev]);
2829 
2830 	dir = ((ev < SPOE_EV_ON_SERVER_SESS) ? SMP_OPT_DIR_REQ : SMP_OPT_DIR_RES);
2831 
2832 	if (LIST_ISEMPTY(&(ctx->events[ev])))
2833 		return 1;
2834 
2835 	ret = spoe_process_messages(s, ctx, &(ctx->events[ev]), dir, SPOE_MSGS_BY_EVENT);
2836 	if (ret && ctx->stats.t_process != -1) {
2837 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
2838 			    " - <EVENT:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu %u/%u\n",
2839 			    (int)now.tv_sec, (int)now.tv_usec, agent->id,
2840 			    __FUNCTION__, s, spoe_event_str[ev], s->uniq_id, ctx->status_code,
2841 			    ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2842 			    ctx->stats.t_response, ctx->stats.t_process,
2843 			    agent->counters.idles, agent->counters.applets,
2844 			    agent->counters.nb_sending, agent->counters.nb_waiting,
2845 			    agent->counters.nb_errors, agent->counters.nb_processed,
2846 			    agent->rt[tid].processing, read_freq_ctr(&agent->rt[tid].processing_per_sec));
2847 		if (ctx->status_code || !(conf->agent_fe.options2 & PR_O2_NOLOGNORM))
2848 			send_log(&conf->agent_fe, (!ctx->status_code ? LOG_NOTICE : LOG_WARNING),
2849 				 "SPOE: [%s] <EVENT:%s> sid=%u st=%u %ld/%ld/%ld/%ld/%ld %u/%u %u/%u %llu/%llu\n",
2850 				 agent->id, spoe_event_str[ev], s->uniq_id, ctx->status_code,
2851 				 ctx->stats.t_request, ctx->stats.t_queue, ctx->stats.t_waiting,
2852 				 ctx->stats.t_response, ctx->stats.t_process,
2853 				 agent->counters.idles, agent->counters.applets,
2854 				 agent->counters.nb_sending, agent->counters.nb_waiting,
2855 				 agent->counters.nb_errors, agent->counters.nb_processed);
2856 	}
2857 	return ret;
2858 }
2859 
2860 /***************************************************************************
2861  * Functions that create/destroy SPOE contexts
2862  **************************************************************************/
2863 static int
spoe_acquire_buffer(struct buffer * buf,struct buffer_wait * buffer_wait)2864 spoe_acquire_buffer(struct buffer *buf, struct buffer_wait *buffer_wait)
2865 {
2866 	if (buf->size)
2867 		return 1;
2868 
2869 	if (!LIST_ISEMPTY(&buffer_wait->list)) {
2870 		HA_SPIN_LOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2871 		LIST_DEL(&buffer_wait->list);
2872 		LIST_INIT(&buffer_wait->list);
2873 		HA_SPIN_UNLOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2874 	}
2875 
2876 	if (b_alloc_margin(buf, global.tune.reserved_bufs))
2877 		return 1;
2878 
2879 	HA_SPIN_LOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2880 	LIST_ADDQ(&buffer_wq, &buffer_wait->list);
2881 	HA_SPIN_UNLOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2882 	return 0;
2883 }
2884 
2885 static void
spoe_release_buffer(struct buffer * buf,struct buffer_wait * buffer_wait)2886 spoe_release_buffer(struct buffer *buf, struct buffer_wait *buffer_wait)
2887 {
2888 	if (!LIST_ISEMPTY(&buffer_wait->list)) {
2889 		HA_SPIN_LOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2890 		LIST_DEL(&buffer_wait->list);
2891 		LIST_INIT(&buffer_wait->list);
2892 		HA_SPIN_UNLOCK(BUF_WQ_LOCK, &buffer_wq_lock);
2893 	}
2894 
2895 	/* Release the buffer if needed */
2896 	if (buf->size) {
2897 		b_free(buf);
2898 		offer_buffers(buffer_wait->target, tasks_run_queue);
2899 	}
2900 }
2901 
2902 static int
spoe_wakeup_context(struct spoe_context * ctx)2903 spoe_wakeup_context(struct spoe_context *ctx)
2904 {
2905 	task_wakeup(ctx->strm->task, TASK_WOKEN_MSG);
2906 	return 1;
2907 }
2908 
2909 static struct spoe_context *
spoe_create_context(struct stream * s,struct filter * filter)2910 spoe_create_context(struct stream *s, struct filter *filter)
2911 {
2912 	struct spoe_config  *conf = FLT_CONF(filter);
2913 	struct spoe_context *ctx;
2914 
2915 	ctx = pool_alloc_dirty(pool_head_spoe_ctx);
2916 	if (ctx == NULL) {
2917 		return NULL;
2918 	}
2919 	memset(ctx, 0, sizeof(*ctx));
2920 	ctx->filter      = filter;
2921 	ctx->state       = SPOE_CTX_ST_NONE;
2922 	ctx->status_code = SPOE_CTX_ERR_NONE;
2923 	ctx->flags       = 0;
2924 	ctx->events      = conf->agent->events;
2925 	ctx->groups      = &conf->agent->groups;
2926 	ctx->buffer      = BUF_NULL;
2927 	LIST_INIT(&ctx->buffer_wait.list);
2928 	ctx->buffer_wait.target = ctx;
2929 	ctx->buffer_wait.wakeup_cb = (int (*)(void *))spoe_wakeup_context;
2930 	LIST_INIT(&ctx->list);
2931 
2932 	ctx->stream_id   = 0;
2933 	ctx->frame_id    = 1;
2934 	ctx->process_exp = TICK_ETERNITY;
2935 
2936 	tv_zero(&ctx->stats.tv_start);
2937 	tv_zero(&ctx->stats.tv_request);
2938 	tv_zero(&ctx->stats.tv_queue);
2939 	tv_zero(&ctx->stats.tv_wait);
2940 	tv_zero(&ctx->stats.tv_response);
2941 	ctx->stats.t_request  = -1;
2942 	ctx->stats.t_queue    = -1;
2943 	ctx->stats.t_waiting  = -1;
2944 	ctx->stats.t_response = -1;
2945 	ctx->stats.t_process  = -1;
2946 	ctx->stats.t_total    =  0;
2947 
2948 	ctx->strm   = s;
2949 	ctx->state  = SPOE_CTX_ST_READY;
2950 	filter->ctx = ctx;
2951 
2952 	return ctx;
2953 }
2954 
2955 static void
spoe_destroy_context(struct filter * filter)2956 spoe_destroy_context(struct filter *filter)
2957 {
2958 	struct spoe_config  *conf = FLT_CONF(filter);
2959 	struct spoe_context *ctx  = filter->ctx;
2960 
2961 	if (!ctx)
2962 		return;
2963 
2964 	spoe_stop_processing(conf->agent, ctx);
2965 	pool_free(pool_head_spoe_ctx, ctx);
2966 	filter->ctx = NULL;
2967 }
2968 
2969 static void
spoe_reset_context(struct spoe_context * ctx)2970 spoe_reset_context(struct spoe_context *ctx)
2971 {
2972 	ctx->state  = SPOE_CTX_ST_READY;
2973 	ctx->flags &= ~(SPOE_CTX_FL_PROCESS|SPOE_CTX_FL_FRAGMENTED);
2974 
2975 	tv_zero(&ctx->stats.tv_start);
2976 	tv_zero(&ctx->stats.tv_request);
2977 	tv_zero(&ctx->stats.tv_queue);
2978 	tv_zero(&ctx->stats.tv_wait);
2979 	tv_zero(&ctx->stats.tv_response);
2980 	ctx->stats.t_request  = -1;
2981 	ctx->stats.t_queue    = -1;
2982 	ctx->stats.t_waiting  = -1;
2983 	ctx->stats.t_response = -1;
2984 	ctx->stats.t_process  = -1;
2985 	ctx->stats.t_total    =  0;
2986 }
2987 
2988 
2989 /***************************************************************************
2990  * Hooks that manage the filter lifecycle (init/check/deinit)
2991  **************************************************************************/
2992 /* Signal handler: Do a soft stop, wakeup SPOE applet */
2993 static void
spoe_sig_stop(struct sig_handler * sh)2994 spoe_sig_stop(struct sig_handler *sh)
2995 {
2996 	struct proxy *p;
2997 
2998 	p = proxies_list;
2999 	while (p) {
3000 		struct flt_conf *fconf;
3001 
3002 		list_for_each_entry(fconf, &p->filter_configs, list) {
3003 			struct spoe_config *conf;
3004 			struct spoe_agent  *agent;
3005 			struct spoe_appctx *spoe_appctx;
3006 			int i;
3007 
3008 			if (fconf->id != spoe_filter_id)
3009 				continue;
3010 
3011 			conf  = fconf->conf;
3012 			agent = conf->agent;
3013 
3014 			for (i = 0; i < global.nbthread; ++i) {
3015 				HA_SPIN_LOCK(SPOE_APPLET_LOCK, &agent->rt[i].lock);
3016 				list_for_each_entry(spoe_appctx, &agent->rt[i].applets, list)
3017 					spoe_wakeup_appctx(spoe_appctx->owner);
3018 				HA_SPIN_UNLOCK(SPOE_APPLET_LOCK, &agent->rt[i].lock);
3019 			}
3020 		}
3021 		p = p->next;
3022 	}
3023 }
3024 
3025 
3026 /* Initialize the SPOE filter. Returns -1 on error, else 0. */
3027 static int
spoe_init(struct proxy * px,struct flt_conf * fconf)3028 spoe_init(struct proxy *px, struct flt_conf *fconf)
3029 {
3030 	struct spoe_config *conf = fconf->conf;
3031 
3032 	/* conf->agent_fe was already initialized during the config
3033 	 * parsing. Finish initialization. */
3034         conf->agent_fe.last_change = now.tv_sec;
3035         conf->agent_fe.cap = PR_CAP_FE;
3036         conf->agent_fe.mode = PR_MODE_TCP;
3037         conf->agent_fe.maxconn = 0;
3038         conf->agent_fe.options2 |= PR_O2_INDEPSTR;
3039         conf->agent_fe.conn_retries = CONN_RETRIES;
3040         conf->agent_fe.accept = frontend_accept;
3041         conf->agent_fe.srv = NULL;
3042         conf->agent_fe.timeout.client = TICK_ETERNITY;
3043 	conf->agent_fe.default_target = &spoe_applet.obj_type;
3044 	conf->agent_fe.fe_req_ana = AN_REQ_SWITCHING_RULES;
3045 
3046 	if (!sighandler_registered) {
3047 		signal_register_fct(0, spoe_sig_stop, 0);
3048 		sighandler_registered = 1;
3049 	}
3050 
3051 	fconf->flags |= FLT_CFG_FL_HTX;
3052 	return 0;
3053 }
3054 
3055 /* Free ressources allocated by the SPOE filter. */
3056 static void
spoe_deinit(struct proxy * px,struct flt_conf * fconf)3057 spoe_deinit(struct proxy *px, struct flt_conf *fconf)
3058 {
3059 	struct spoe_config *conf = fconf->conf;
3060 
3061 	if (conf) {
3062 		struct spoe_agent *agent = conf->agent;
3063 
3064 		spoe_release_agent(agent);
3065 		free(conf->id);
3066 		free(conf);
3067 	}
3068 	fconf->conf = NULL;
3069 }
3070 
3071 /* Check configuration of a SPOE filter for a specified proxy.
3072  * Return 1 on error, else 0. */
3073 static int
spoe_check(struct proxy * px,struct flt_conf * fconf)3074 spoe_check(struct proxy *px, struct flt_conf *fconf)
3075 {
3076 	struct flt_conf    *f;
3077 	struct spoe_config *conf = fconf->conf;
3078 	struct proxy       *target;
3079 	int i;
3080 
3081 	/* Check all SPOE filters for proxy <px> to be sure all SPOE agent names
3082 	 * are uniq */
3083 	list_for_each_entry(f, &px->filter_configs, list) {
3084 		struct spoe_config *c = f->conf;
3085 
3086 		/* This is not an SPOE filter */
3087 		if (f->id != spoe_filter_id)
3088 			continue;
3089 		/* This is the current SPOE filter */
3090 		if (f == fconf)
3091 			continue;
3092 
3093 		/* Check engine Id. It should be uniq */
3094 		if (!strcmp(conf->id, c->id)) {
3095 			ha_alert("Proxy %s : duplicated name for SPOE engine '%s'.\n",
3096 				 px->id, conf->id);
3097 			return 1;
3098 		}
3099 	}
3100 
3101 	target = proxy_be_by_name(conf->agent->b.name);
3102 	if (target == NULL) {
3103 		ha_alert("Proxy %s : unknown backend '%s' used by SPOE agent '%s'"
3104 			 " declared at %s:%d.\n",
3105 			 px->id, conf->agent->b.name, conf->agent->id,
3106 			 conf->agent->conf.file, conf->agent->conf.line);
3107 		return 1;
3108 	}
3109 	if (target->mode != PR_MODE_TCP) {
3110 		ha_alert("Proxy %s : backend '%s' used by SPOE agent '%s' declared"
3111 			 " at %s:%d does not support HTTP mode.\n",
3112 			 px->id, target->id, conf->agent->id,
3113 			 conf->agent->conf.file, conf->agent->conf.line);
3114 		return 1;
3115 	}
3116 
3117 	if (px->bind_proc & ~target->bind_proc) {
3118 		ha_alert("Proxy %s : backend '%s' used by SPOE agent '%s' declared"
3119 			 " at %s:%d does not cover all of its processes.\n",
3120 			 px->id, target->id, conf->agent->id,
3121 			 conf->agent->conf.file, conf->agent->conf.line);
3122 		return 1;
3123 	}
3124 
3125 	if ((conf->agent->rt = calloc(global.nbthread, sizeof(*conf->agent->rt))) == NULL) {
3126 		ha_alert("Proxy %s : out of memory initializing SPOE agent '%s' declared at %s:%d.\n",
3127 			 px->id, conf->agent->id, conf->agent->conf.file, conf->agent->conf.line);
3128 		return 1;
3129 	}
3130 	for (i = 0; i < global.nbthread; ++i) {
3131 		conf->agent->rt[i].engine_id    = NULL;
3132 		conf->agent->rt[i].frame_size   = conf->agent->max_frame_size;
3133 		conf->agent->rt[i].processing   = 0;
3134 		LIST_INIT(&conf->agent->rt[i].applets);
3135 		LIST_INIT(&conf->agent->rt[i].sending_queue);
3136 		LIST_INIT(&conf->agent->rt[i].waiting_queue);
3137 		HA_SPIN_INIT(&conf->agent->rt[i].lock);
3138 	}
3139 
3140 	free(conf->agent->b.name);
3141 	conf->agent->b.name = NULL;
3142 	conf->agent->b.be = target;
3143 	return 0;
3144 }
3145 
3146 /* Initializes the SPOE filter for a proxy for a specific thread.
3147  * Returns a negative value if an error occurs. */
3148 static int
spoe_init_per_thread(struct proxy * p,struct flt_conf * fconf)3149 spoe_init_per_thread(struct proxy *p, struct flt_conf *fconf)
3150 {
3151 	struct spoe_config *conf = fconf->conf;
3152 	struct spoe_agent *agent = conf->agent;
3153 
3154 	agent->rt[tid].engine_id = generate_pseudo_uuid();
3155 	if (agent->rt[tid].engine_id == NULL)
3156 		return -1;
3157 	return 0;
3158 }
3159 
3160 /**************************************************************************
3161  * Hooks attached to a stream
3162  *************************************************************************/
3163 /* Called when a filter instance is created and attach to a stream. It creates
3164  * the context that will be used to process this stream. */
3165 static int
spoe_start(struct stream * s,struct filter * filter)3166 spoe_start(struct stream *s, struct filter *filter)
3167 {
3168 	struct spoe_config  *conf  = FLT_CONF(filter);
3169 	struct spoe_agent   *agent = conf->agent;
3170 	struct spoe_context *ctx;
3171 
3172 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p\n",
3173 		    (int)now.tv_sec, (int)now.tv_usec, agent->id,
3174 		    __FUNCTION__, s);
3175 
3176 	if ((ctx = spoe_create_context(s, filter)) == NULL) {
3177 		SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
3178 			    " - failed to create SPOE context\n",
3179 			    (int)now.tv_sec, (int)now.tv_usec, agent->id,
3180 			    __FUNCTION__, s);
3181 		send_log(&conf->agent_fe, LOG_EMERG,
3182 			 "SPOE: [%s] failed to create SPOE context\n",
3183 			 agent->id);
3184 		return 0;
3185 	}
3186 
3187 	if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_REQ_FE]))
3188 		filter->pre_analyzers |= AN_REQ_INSPECT_FE;
3189 
3190 	if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_REQ_BE]))
3191 		filter->pre_analyzers |= AN_REQ_INSPECT_BE;
3192 
3193 	if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_TCP_RSP]))
3194 		filter->pre_analyzers |= AN_RES_INSPECT;
3195 
3196 	if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_HTTP_REQ_FE]))
3197 		filter->pre_analyzers |= AN_REQ_HTTP_PROCESS_FE;
3198 
3199 	if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_HTTP_REQ_BE]))
3200 		filter->pre_analyzers |= AN_REQ_HTTP_PROCESS_BE;
3201 
3202 	if (!LIST_ISEMPTY(&ctx->events[SPOE_EV_ON_HTTP_RSP]))
3203 		filter->pre_analyzers |= AN_RES_HTTP_PROCESS_FE;
3204 
3205 	return 1;
3206 }
3207 
3208 /* Called when a filter instance is detached from a stream. It release the
3209  * attached SPOE context. */
3210 static void
spoe_stop(struct stream * s,struct filter * filter)3211 spoe_stop(struct stream *s, struct filter *filter)
3212 {
3213 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p\n",
3214 		    (int)now.tv_sec, (int)now.tv_usec,
3215 		    ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3216 		    __FUNCTION__, s);
3217 	spoe_destroy_context(filter);
3218 }
3219 
3220 
3221 /*
3222  * Called when the stream is woken up because of expired timer.
3223  */
3224 static void
spoe_check_timeouts(struct stream * s,struct filter * filter)3225 spoe_check_timeouts(struct stream *s, struct filter *filter)
3226 {
3227 	struct spoe_context *ctx = filter->ctx;
3228 
3229 	if (tick_is_expired(ctx->process_exp, now_ms))
3230 		s->pending_events |= TASK_WOKEN_MSG;
3231 }
3232 
3233 /* Called when we are ready to filter data on a channel */
3234 static int
spoe_start_analyze(struct stream * s,struct filter * filter,struct channel * chn)3235 spoe_start_analyze(struct stream *s, struct filter *filter, struct channel *chn)
3236 {
3237 	struct spoe_context *ctx = filter->ctx;
3238 	int                  ret = 1;
3239 
3240 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p - ctx-state=%s"
3241 		    " - ctx-flags=0x%08x\n",
3242 		    (int)now.tv_sec, (int)now.tv_usec,
3243 		    ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3244 		    __FUNCTION__, s, spoe_ctx_state_str[ctx->state], ctx->flags);
3245 
3246 	if (ctx->state == SPOE_CTX_ST_NONE)
3247 		goto out;
3248 
3249 	if (!(chn->flags & CF_ISRESP)) {
3250 		if (filter->pre_analyzers & AN_REQ_INSPECT_FE)
3251 			chn->analysers |= AN_REQ_INSPECT_FE;
3252 		if (filter->pre_analyzers & AN_REQ_INSPECT_BE)
3253 			chn->analysers |= AN_REQ_INSPECT_BE;
3254 
3255 		if (ctx->flags & SPOE_CTX_FL_CLI_CONNECTED)
3256 			goto out;
3257 
3258 		ctx->stream_id = s->uniq_id;
3259 		ret = spoe_process_event(s, ctx, SPOE_EV_ON_CLIENT_SESS);
3260 		if (!ret)
3261 			goto out;
3262 		ctx->flags |= SPOE_CTX_FL_CLI_CONNECTED;
3263 	}
3264 	else {
3265 		if (filter->pre_analyzers & AN_RES_INSPECT)
3266 			chn->analysers |= AN_RES_INSPECT;
3267 
3268 		if (ctx->flags & SPOE_CTX_FL_SRV_CONNECTED)
3269 			goto out;
3270 
3271 		ret = spoe_process_event(s, ctx, SPOE_EV_ON_SERVER_SESS);
3272 		if (!ret) {
3273 			channel_dont_read(chn);
3274 			channel_dont_close(chn);
3275 			goto out;
3276 		}
3277 		ctx->flags |= SPOE_CTX_FL_SRV_CONNECTED;
3278 	}
3279 
3280   out:
3281 	return ret;
3282 }
3283 
3284 /* Called before a processing happens on a given channel */
3285 static int
spoe_chn_pre_analyze(struct stream * s,struct filter * filter,struct channel * chn,unsigned an_bit)3286 spoe_chn_pre_analyze(struct stream *s, struct filter *filter,
3287 		     struct channel *chn, unsigned an_bit)
3288 {
3289 	struct spoe_context *ctx = filter->ctx;
3290 	int                  ret = 1;
3291 
3292 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p - ctx-state=%s"
3293 		    " - ctx-flags=0x%08x - ana=0x%08x\n",
3294 		    (int)now.tv_sec, (int)now.tv_usec,
3295 		    ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3296 		    __FUNCTION__, s, spoe_ctx_state_str[ctx->state],
3297 		    ctx->flags, an_bit);
3298 
3299 	if (ctx->state == SPOE_CTX_ST_NONE)
3300 		goto out;
3301 
3302 	switch (an_bit) {
3303 		case AN_REQ_INSPECT_FE:
3304 			ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_REQ_FE);
3305 			break;
3306 		case AN_REQ_INSPECT_BE:
3307 			ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_REQ_BE);
3308 			break;
3309 		case AN_RES_INSPECT:
3310 			ret = spoe_process_event(s, ctx, SPOE_EV_ON_TCP_RSP);
3311 			break;
3312 		case AN_REQ_HTTP_PROCESS_FE:
3313 			ret = spoe_process_event(s, ctx, SPOE_EV_ON_HTTP_REQ_FE);
3314 			break;
3315 		case AN_REQ_HTTP_PROCESS_BE:
3316 			ret = spoe_process_event(s, ctx, SPOE_EV_ON_HTTP_REQ_BE);
3317 			break;
3318 		case AN_RES_HTTP_PROCESS_FE:
3319 			ret = spoe_process_event(s, ctx, SPOE_EV_ON_HTTP_RSP);
3320 			break;
3321 	}
3322 
3323   out:
3324 	if (!ret && (chn->flags & CF_ISRESP)) {
3325                 channel_dont_read(chn);
3326                 channel_dont_close(chn);
3327 	}
3328 	return ret;
3329 }
3330 
3331 /* Called when the filtering on the channel ends. */
3332 static int
spoe_end_analyze(struct stream * s,struct filter * filter,struct channel * chn)3333 spoe_end_analyze(struct stream *s, struct filter *filter, struct channel *chn)
3334 {
3335 	struct spoe_context *ctx = filter->ctx;
3336 
3337 	SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p - ctx-state=%s"
3338 		    " - ctx-flags=0x%08x\n",
3339 		    (int)now.tv_sec, (int)now.tv_usec,
3340 		    ((struct spoe_config *)FLT_CONF(filter))->agent->id,
3341 		    __FUNCTION__, s, spoe_ctx_state_str[ctx->state], ctx->flags);
3342 
3343 	if (!(ctx->flags & SPOE_CTX_FL_PROCESS)) {
3344 		spoe_reset_context(ctx);
3345 	}
3346 
3347 	return 1;
3348 }
3349 
3350 /********************************************************************
3351  * Functions that manage the filter initialization
3352  ********************************************************************/
3353 struct flt_ops spoe_ops = {
3354 	/* Manage SPOE filter, called for each filter declaration */
3355 	.init   = spoe_init,
3356 	.deinit = spoe_deinit,
3357 	.check  = spoe_check,
3358 	.init_per_thread = spoe_init_per_thread,
3359 
3360 	/* Handle start/stop of SPOE */
3361 	.attach         = spoe_start,
3362 	.detach         = spoe_stop,
3363 	.check_timeouts = spoe_check_timeouts,
3364 
3365 	/* Handle channels activity */
3366 	.channel_start_analyze = spoe_start_analyze,
3367 	.channel_pre_analyze   = spoe_chn_pre_analyze,
3368 	.channel_end_analyze   = spoe_end_analyze,
3369 };
3370 
3371 
3372 static int
cfg_parse_spoe_agent(const char * file,int linenum,char ** args,int kwm)3373 cfg_parse_spoe_agent(const char *file, int linenum, char **args, int kwm)
3374 {
3375 	const char *err;
3376 	int         i, err_code = 0;
3377 
3378 	if ((cfg_scope == NULL && curengine != NULL) ||
3379 	    (cfg_scope != NULL && curengine == NULL) ||
3380 	    (curengine != NULL && cfg_scope != NULL && strcmp(curengine, cfg_scope)))
3381 		goto out;
3382 
3383 	if (!strcmp(args[0], "spoe-agent")) { /* new spoe-agent section */
3384 		if (!*args[1]) {
3385 			ha_alert("parsing [%s:%d] : missing name for spoe-agent section.\n",
3386 				 file, linenum);
3387 			err_code |= ERR_ALERT | ERR_ABORT;
3388 			goto out;
3389 		}
3390 		if (alertif_too_many_args(1, file, linenum, args, &err_code)) {
3391 			err_code |= ERR_ABORT;
3392 			goto out;
3393 		}
3394 
3395 		err = invalid_char(args[1]);
3396 		if (err) {
3397 			ha_alert("parsing [%s:%d] : character '%c' is not permitted in '%s' name '%s'.\n",
3398 				 file, linenum, *err, args[0], args[1]);
3399 			err_code |= ERR_ALERT | ERR_ABORT;
3400 			goto out;
3401 		}
3402 
3403 		if (curagent != NULL) {
3404 			ha_alert("parsing [%s:%d] : another spoe-agent section previously defined.\n",
3405 				 file, linenum);
3406 			err_code |= ERR_ALERT | ERR_ABORT;
3407 			goto out;
3408 		}
3409 		if ((curagent = calloc(1, sizeof(*curagent))) == NULL) {
3410 			ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3411 			err_code |= ERR_ALERT | ERR_ABORT;
3412 			goto out;
3413 		}
3414 
3415 		curagent->id              = strdup(args[1]);
3416 
3417 		curagent->conf.file       = strdup(file);
3418 		curagent->conf.line       = linenum;
3419 
3420 		curagent->timeout.hello      = TICK_ETERNITY;
3421 		curagent->timeout.idle       = TICK_ETERNITY;
3422 		curagent->timeout.processing = TICK_ETERNITY;
3423 
3424 		curagent->var_pfx        = NULL;
3425 		curagent->var_on_error   = NULL;
3426 		curagent->var_t_process  = NULL;
3427 		curagent->var_t_total    = NULL;
3428 		curagent->flags          = (SPOE_FL_ASYNC | SPOE_FL_PIPELINING | SPOE_FL_SND_FRAGMENTATION);
3429 		curagent->cps_max        = 0;
3430 		curagent->eps_max        = 0;
3431 		curagent->max_frame_size = MAX_FRAME_SIZE;
3432 		curagent->max_fpa        = 20;
3433 
3434 		for (i = 0; i < SPOE_EV_EVENTS; ++i)
3435 			LIST_INIT(&curagent->events[i]);
3436 		LIST_INIT(&curagent->groups);
3437 		LIST_INIT(&curagent->messages);
3438 	}
3439 	else if (!strcmp(args[0], "use-backend")) {
3440 		if (!*args[1]) {
3441 			ha_alert("parsing [%s:%d] : '%s' expects a backend name.\n",
3442 				 file, linenum, args[0]);
3443 			err_code |= ERR_ALERT | ERR_FATAL;
3444 			goto out;
3445 		}
3446 		if (alertif_too_many_args(1, file, linenum, args, &err_code))
3447 			goto out;
3448 		free(curagent->b.name);
3449 		curagent->b.name = strdup(args[1]);
3450 	}
3451 	else if (!strcmp(args[0], "messages")) {
3452 		int cur_arg = 1;
3453 		while (*args[cur_arg]) {
3454 			struct spoe_placeholder *ph = NULL;
3455 
3456 			list_for_each_entry(ph, &curmphs, list) {
3457 				if (!strcmp(ph->id, args[cur_arg])) {
3458 					ha_alert("parsing [%s:%d]: spoe-message '%s' already used.\n",
3459 						 file, linenum, args[cur_arg]);
3460 					err_code |= ERR_ALERT | ERR_FATAL;
3461 					goto out;
3462 				}
3463 			}
3464 
3465 			if ((ph = calloc(1, sizeof(*ph))) == NULL) {
3466 				ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3467 				err_code |= ERR_ALERT | ERR_ABORT;
3468 				goto out;
3469 			}
3470 			ph->id = strdup(args[cur_arg]);
3471 			LIST_ADDQ(&curmphs, &ph->list);
3472 			cur_arg++;
3473 		}
3474 	}
3475 	else if (!strcmp(args[0], "groups")) {
3476 		int cur_arg = 1;
3477 		while (*args[cur_arg]) {
3478 			struct spoe_placeholder *ph = NULL;
3479 
3480 			list_for_each_entry(ph, &curgphs, list) {
3481 				if (!strcmp(ph->id, args[cur_arg])) {
3482 					ha_alert("parsing [%s:%d]: spoe-group '%s' already used.\n",
3483 						 file, linenum, args[cur_arg]);
3484 					err_code |= ERR_ALERT | ERR_FATAL;
3485 					goto out;
3486 				}
3487 			}
3488 
3489 			if ((ph = calloc(1, sizeof(*ph))) == NULL) {
3490 				ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3491 				err_code |= ERR_ALERT | ERR_ABORT;
3492 				goto out;
3493 			}
3494 			ph->id = strdup(args[cur_arg]);
3495 			LIST_ADDQ(&curgphs, &ph->list);
3496 			cur_arg++;
3497 		}
3498 	}
3499 	else if (!strcmp(args[0], "timeout")) {
3500 		unsigned int *tv = NULL;
3501 		const char   *res;
3502 		unsigned      timeout;
3503 
3504 		if (!*args[1]) {
3505 			ha_alert("parsing [%s:%d] : 'timeout' expects 'hello', 'idle' and 'processing'.\n",
3506 				 file, linenum);
3507 			err_code |= ERR_ALERT | ERR_FATAL;
3508 			goto out;
3509 		}
3510 		if (alertif_too_many_args(2, file, linenum, args, &err_code))
3511 			goto out;
3512 		if (!strcmp(args[1], "hello"))
3513 			tv = &curagent->timeout.hello;
3514 		else if (!strcmp(args[1], "idle"))
3515 			tv = &curagent->timeout.idle;
3516 		else if (!strcmp(args[1], "processing"))
3517 			tv = &curagent->timeout.processing;
3518 		else {
3519 			ha_alert("parsing [%s:%d] : 'timeout' supports 'hello', 'idle' or 'processing' (got %s).\n",
3520 				 file, linenum, args[1]);
3521 			err_code |= ERR_ALERT | ERR_FATAL;
3522 			goto out;
3523 		}
3524 		if (!*args[2]) {
3525 			ha_alert("parsing [%s:%d] : 'timeout %s' expects an integer value (in milliseconds).\n",
3526 				 file, linenum, args[1]);
3527 			err_code |= ERR_ALERT | ERR_FATAL;
3528 			goto out;
3529 		}
3530 		res = parse_time_err(args[2], &timeout, TIME_UNIT_MS);
3531 		if (res == PARSE_TIME_OVER) {
3532 			ha_alert("parsing [%s:%d]: timer overflow in argument <%s> to <%s %s>, maximum value is 2147483647 ms (~24.8 days).\n",
3533 				 file, linenum, args[2], args[0], args[1]);
3534 			err_code |= ERR_ALERT | ERR_FATAL;
3535 			goto out;
3536 		}
3537 		else if (res == PARSE_TIME_UNDER) {
3538 			ha_alert("parsing [%s:%d]: timer underflow in argument <%s> to <%s %s>, minimum non-null value is 1 ms.\n",
3539 				 file, linenum, args[2], args[0], args[1]);
3540 			err_code |= ERR_ALERT | ERR_FATAL;
3541 			goto out;
3542 		}
3543 		else if (res) {
3544 			ha_alert("parsing [%s:%d] : unexpected character '%c' in 'timeout %s'.\n",
3545 				 file, linenum, *res, args[1]);
3546 			err_code |= ERR_ALERT | ERR_FATAL;
3547 			goto out;
3548 		}
3549 		*tv = MS_TO_TICKS(timeout);
3550 	}
3551 	else if (!strcmp(args[0], "option")) {
3552 		if (!*args[1]) {
3553                         ha_alert("parsing [%s:%d]: '%s' expects an option name.\n",
3554 				 file, linenum, args[0]);
3555                         err_code |= ERR_ALERT | ERR_FATAL;
3556                         goto out;
3557                 }
3558 
3559 		if (!strcmp(args[1], "pipelining")) {
3560 			if (alertif_too_many_args(1, file, linenum, args, &err_code))
3561 				goto out;
3562 			if (kwm == 1)
3563 				curagent->flags &= ~SPOE_FL_PIPELINING;
3564 			else
3565 				curagent->flags |= SPOE_FL_PIPELINING;
3566 			goto out;
3567 		}
3568 		else if (!strcmp(args[1], "async")) {
3569 			if (alertif_too_many_args(1, file, linenum, args, &err_code))
3570 				goto out;
3571 			if (kwm == 1)
3572 				curagent->flags &= ~SPOE_FL_ASYNC;
3573 			else
3574 				curagent->flags |= SPOE_FL_ASYNC;
3575 			goto out;
3576 		}
3577 		else if (!strcmp(args[1], "send-frag-payload")) {
3578 			if (alertif_too_many_args(1, file, linenum, args, &err_code))
3579 				goto out;
3580 			if (kwm == 1)
3581 				curagent->flags &= ~SPOE_FL_SND_FRAGMENTATION;
3582 			else
3583 				curagent->flags |= SPOE_FL_SND_FRAGMENTATION;
3584 			goto out;
3585 		}
3586 		else if (!strcmp(args[1], "dontlog-normal")) {
3587 			if (alertif_too_many_args(1, file, linenum, args, &err_code))
3588 				goto out;
3589 			if (kwm == 1)
3590 				curpxopts2 &= ~PR_O2_NOLOGNORM;
3591 			else
3592 				curpxopts2 |= PR_O2_NOLOGNORM;
3593 			goto out;
3594 		}
3595 
3596 		/* Following options does not support negation */
3597 		if (kwm == 1) {
3598 			ha_alert("parsing [%s:%d]: negation is not supported for option '%s'.\n",
3599 				 file, linenum, args[1]);
3600 			err_code |= ERR_ALERT | ERR_FATAL;
3601 			goto out;
3602 		}
3603 
3604 		if (!strcmp(args[1], "var-prefix")) {
3605 			char *tmp;
3606 
3607 			if (!*args[2]) {
3608 				ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3609 					 file, linenum, args[0],
3610 					 args[1]);
3611 				err_code |= ERR_ALERT | ERR_FATAL;
3612 				goto out;
3613 			}
3614 			if (alertif_too_many_args(2, file, linenum, args, &err_code))
3615 				goto out;
3616 			tmp = args[2];
3617 			while (*tmp) {
3618 				if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3619 					ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3620 						 file, linenum, args[0], args[1]);
3621 					err_code |= ERR_ALERT | ERR_FATAL;
3622 					goto out;
3623 				}
3624 				tmp++;
3625 			}
3626 			curagent->var_pfx = strdup(args[2]);
3627 		}
3628 		else if (!strcmp(args[1], "force-set-var")) {
3629 			if (alertif_too_many_args(1, file, linenum, args, &err_code))
3630 				goto out;
3631 			curagent->flags |= SPOE_FL_FORCE_SET_VAR;
3632 		}
3633 		else if (!strcmp(args[1], "continue-on-error")) {
3634 			if (alertif_too_many_args(1, file, linenum, args, &err_code))
3635 				goto out;
3636 			curagent->flags |= SPOE_FL_CONT_ON_ERR;
3637 		}
3638 		else if (!strcmp(args[1], "set-on-error")) {
3639 			char *tmp;
3640 
3641 			if (!*args[2]) {
3642 				ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3643 					 file, linenum, args[0],
3644 					 args[1]);
3645 				err_code |= ERR_ALERT | ERR_FATAL;
3646 				goto out;
3647 			}
3648 			if (alertif_too_many_args(2, file, linenum, args, &err_code))
3649 				goto out;
3650 			tmp = args[2];
3651 			while (*tmp) {
3652 				if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3653 					ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3654 						 file, linenum, args[0], args[1]);
3655 					err_code |= ERR_ALERT | ERR_FATAL;
3656 					goto out;
3657 				}
3658 				tmp++;
3659 			}
3660 			curagent->var_on_error = strdup(args[2]);
3661 		}
3662 		else if (!strcmp(args[1], "set-process-time")) {
3663 			char *tmp;
3664 
3665 			if (!*args[2]) {
3666 				ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3667 					 file, linenum, args[0],
3668 					 args[1]);
3669 				err_code |= ERR_ALERT | ERR_FATAL;
3670 				goto out;
3671 			}
3672 			if (alertif_too_many_args(2, file, linenum, args, &err_code))
3673 				goto out;
3674 			tmp = args[2];
3675 			while (*tmp) {
3676 				if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3677 					ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3678 						 file, linenum, args[0], args[1]);
3679 					err_code |= ERR_ALERT | ERR_FATAL;
3680 					goto out;
3681 				}
3682 				tmp++;
3683 			}
3684 			curagent->var_t_process = strdup(args[2]);
3685 		}
3686 		else if (!strcmp(args[1], "set-total-time")) {
3687 			char *tmp;
3688 
3689 			if (!*args[2]) {
3690 				ha_alert("parsing [%s:%d]: '%s %s' expects a value.\n",
3691 					 file, linenum, args[0],
3692 					 args[1]);
3693 				err_code |= ERR_ALERT | ERR_FATAL;
3694 				goto out;
3695 			}
3696 			if (alertif_too_many_args(2, file, linenum, args, &err_code))
3697 				goto out;
3698 			tmp = args[2];
3699 			while (*tmp) {
3700 				if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
3701 					ha_alert("parsing [%s:%d]: '%s %s' only supports [a-zA-Z0-9_.] chars.\n",
3702 						 file, linenum, args[0], args[1]);
3703 					err_code |= ERR_ALERT | ERR_FATAL;
3704 					goto out;
3705 				}
3706 				tmp++;
3707 			}
3708 			curagent->var_t_total = strdup(args[2]);
3709 		}
3710 		else {
3711 			ha_alert("parsing [%s:%d]: option '%s' is not supported.\n",
3712 				 file, linenum, args[1]);
3713 			err_code |= ERR_ALERT | ERR_FATAL;
3714 			goto out;
3715 		}
3716 	}
3717 	else if (!strcmp(args[0], "maxconnrate")) {
3718 		if (!*args[1]) {
3719 			ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3720 				 file, linenum, args[0]);
3721                         err_code |= ERR_ALERT | ERR_FATAL;
3722                         goto out;
3723                 }
3724 		if (alertif_too_many_args(1, file, linenum, args, &err_code))
3725 			goto out;
3726 		curagent->cps_max = atol(args[1]);
3727 	}
3728 	else if (!strcmp(args[0], "maxerrrate")) {
3729 		if (!*args[1]) {
3730 			ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3731 				 file, linenum, args[0]);
3732                         err_code |= ERR_ALERT | ERR_FATAL;
3733                         goto out;
3734                 }
3735 		if (alertif_too_many_args(1, file, linenum, args, &err_code))
3736 			goto out;
3737 		curagent->eps_max = atol(args[1]);
3738 	}
3739 	else if (!strcmp(args[0], "max-frame-size")) {
3740 		if (!*args[1]) {
3741 			ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3742 				 file, linenum, args[0]);
3743                         err_code |= ERR_ALERT | ERR_FATAL;
3744                         goto out;
3745                 }
3746 		if (alertif_too_many_args(1, file, linenum, args, &err_code))
3747 			goto out;
3748 		curagent->max_frame_size = atol(args[1]);
3749 		if (curagent->max_frame_size < MIN_FRAME_SIZE ||
3750 		    curagent->max_frame_size > MAX_FRAME_SIZE) {
3751 			ha_alert("parsing [%s:%d] : '%s' expects a positive integer argument in the range [%d, %d].\n",
3752 				 file, linenum, args[0], MIN_FRAME_SIZE, MAX_FRAME_SIZE);
3753 			err_code |= ERR_ALERT | ERR_FATAL;
3754 			goto out;
3755 		}
3756 	}
3757 	else if (!strcmp(args[0], "max-waiting-frames")) {
3758 		if (!*args[1]) {
3759 			ha_alert("parsing [%s:%d] : '%s' expects an integer argument.\n",
3760 				 file, linenum, args[0]);
3761                         err_code |= ERR_ALERT | ERR_FATAL;
3762                         goto out;
3763                 }
3764 		if (alertif_too_many_args(1, file, linenum, args, &err_code))
3765 			goto out;
3766 		curagent->max_fpa = atol(args[1]);
3767 		if (curagent->max_fpa < 1) {
3768 			ha_alert("parsing [%s:%d] : '%s' expects a positive integer argument.\n",
3769 				 file, linenum, args[0]);
3770 			err_code |= ERR_ALERT | ERR_FATAL;
3771 			goto out;
3772 		}
3773 	}
3774 	else if (!strcmp(args[0], "register-var-names")) {
3775 		int   cur_arg;
3776 
3777 		if (!*args[1]) {
3778 			ha_alert("parsing [%s:%d] : '%s' expects one or more variable names.\n",
3779 				 file, linenum, args[0]);
3780                         err_code |= ERR_ALERT | ERR_FATAL;
3781                         goto out;
3782                 }
3783 		cur_arg = 1;
3784 		while (*args[cur_arg]) {
3785 			struct spoe_var_placeholder *vph;
3786 
3787 			if ((vph = calloc(1, sizeof(*vph))) == NULL) {
3788 				ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3789 				err_code |= ERR_ALERT | ERR_ABORT;
3790 				goto out;
3791 			}
3792 			if ((vph->name  = strdup(args[cur_arg])) == NULL) {
3793 				free(vph);
3794 				ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3795 				err_code |= ERR_ALERT | ERR_ABORT;
3796 				goto out;
3797 			}
3798 			LIST_ADDQ(&curvars, &vph->list);
3799 			cur_arg++;
3800 		}
3801 	}
3802 	else if (!strcmp(args[0], "log")) {
3803 		char *errmsg = NULL;
3804 
3805 		if (!parse_logsrv(args, &curlogsrvs, (kwm == 1), &errmsg)) {
3806 			ha_alert("parsing [%s:%d] : %s : %s\n", file, linenum, args[0], errmsg);
3807 			err_code |= ERR_ALERT | ERR_FATAL;
3808 			goto out;
3809 		}
3810 	}
3811 	else if (*args[0]) {
3812 		ha_alert("parsing [%s:%d] : unknown keyword '%s' in spoe-agent section.\n",
3813 			 file, linenum, args[0]);
3814 		err_code |= ERR_ALERT | ERR_FATAL;
3815 		goto out;
3816 	}
3817  out:
3818 	return err_code;
3819 }
3820 static int
cfg_parse_spoe_group(const char * file,int linenum,char ** args,int kwm)3821 cfg_parse_spoe_group(const char *file, int linenum, char **args, int kwm)
3822 {
3823 	struct spoe_group *grp;
3824 	const char        *err;
3825 	int                err_code = 0;
3826 
3827 	if ((cfg_scope == NULL && curengine != NULL) ||
3828 	    (cfg_scope != NULL && curengine == NULL) ||
3829 	    (curengine != NULL && cfg_scope != NULL && strcmp(curengine, cfg_scope)))
3830 		goto out;
3831 
3832 	if (!strcmp(args[0], "spoe-group")) { /* new spoe-group section */
3833 		if (!*args[1]) {
3834 			ha_alert("parsing [%s:%d] : missing name for spoe-group section.\n",
3835 				 file, linenum);
3836 			err_code |= ERR_ALERT | ERR_ABORT;
3837 			goto out;
3838 		}
3839 		if (alertif_too_many_args(1, file, linenum, args, &err_code)) {
3840 			err_code |= ERR_ABORT;
3841 			goto out;
3842 		}
3843 
3844 		err = invalid_char(args[1]);
3845 		if (err) {
3846 			ha_alert("parsing [%s:%d] : character '%c' is not permitted in '%s' name '%s'.\n",
3847 				 file, linenum, *err, args[0], args[1]);
3848 			err_code |= ERR_ALERT | ERR_ABORT;
3849 			goto out;
3850 		}
3851 
3852 		list_for_each_entry(grp, &curgrps, list) {
3853 			if (!strcmp(grp->id, args[1])) {
3854 				ha_alert("parsing [%s:%d]: spoe-group section '%s' has the same"
3855 					 " name as another one declared at %s:%d.\n",
3856 					 file, linenum, args[1], grp->conf.file, grp->conf.line);
3857 				err_code |= ERR_ALERT | ERR_FATAL;
3858 				goto out;
3859 			}
3860 		}
3861 
3862 		if ((curgrp = calloc(1, sizeof(*curgrp))) == NULL) {
3863 			ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3864 			err_code |= ERR_ALERT | ERR_ABORT;
3865 			goto out;
3866 		}
3867 
3868 		curgrp->id        = strdup(args[1]);
3869 		curgrp->conf.file = strdup(file);
3870 		curgrp->conf.line = linenum;
3871 		LIST_INIT(&curgrp->phs);
3872 		LIST_INIT(&curgrp->messages);
3873 		LIST_ADDQ(&curgrps, &curgrp->list);
3874 	}
3875 	else if (!strcmp(args[0], "messages")) {
3876 		int cur_arg = 1;
3877 		while (*args[cur_arg]) {
3878 			struct spoe_placeholder *ph = NULL;
3879 
3880 			list_for_each_entry(ph, &curgrp->phs, list) {
3881 				if (!strcmp(ph->id, args[cur_arg])) {
3882 					ha_alert("parsing [%s:%d]: spoe-message '%s' already used.\n",
3883 						 file, linenum, args[cur_arg]);
3884 					err_code |= ERR_ALERT | ERR_FATAL;
3885 					goto out;
3886 				}
3887 			}
3888 
3889 			if ((ph = calloc(1, sizeof(*ph))) == NULL) {
3890 				ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3891 				err_code |= ERR_ALERT | ERR_ABORT;
3892 				goto out;
3893 			}
3894 			ph->id = strdup(args[cur_arg]);
3895 			LIST_ADDQ(&curgrp->phs, &ph->list);
3896 			cur_arg++;
3897 		}
3898 	}
3899 	else if (*args[0]) {
3900 		ha_alert("parsing [%s:%d] : unknown keyword '%s' in spoe-group section.\n",
3901 			 file, linenum, args[0]);
3902 		err_code |= ERR_ALERT | ERR_FATAL;
3903 		goto out;
3904 	}
3905  out:
3906 	return err_code;
3907 }
3908 
3909 static int
cfg_parse_spoe_message(const char * file,int linenum,char ** args,int kwm)3910 cfg_parse_spoe_message(const char *file, int linenum, char **args, int kwm)
3911 {
3912 	struct spoe_message *msg;
3913 	struct spoe_arg     *arg;
3914 	const char          *err;
3915 	char                *errmsg   = NULL;
3916 	int                  err_code = 0;
3917 
3918 	if ((cfg_scope == NULL && curengine != NULL) ||
3919 	    (cfg_scope != NULL && curengine == NULL) ||
3920 	    (curengine != NULL && cfg_scope != NULL && strcmp(curengine, cfg_scope)))
3921 		goto out;
3922 
3923 	if (!strcmp(args[0], "spoe-message")) { /* new spoe-message section */
3924 		if (!*args[1]) {
3925 			ha_alert("parsing [%s:%d] : missing name for spoe-message section.\n",
3926 				 file, linenum);
3927 			err_code |= ERR_ALERT | ERR_ABORT;
3928 			goto out;
3929 		}
3930 		if (alertif_too_many_args(1, file, linenum, args, &err_code)) {
3931 			err_code |= ERR_ABORT;
3932 			goto out;
3933 		}
3934 
3935 		err = invalid_char(args[1]);
3936 		if (err) {
3937 			ha_alert("parsing [%s:%d] : character '%c' is not permitted in '%s' name '%s'.\n",
3938 				 file, linenum, *err, args[0], args[1]);
3939 			err_code |= ERR_ALERT | ERR_ABORT;
3940 			goto out;
3941 		}
3942 
3943 		list_for_each_entry(msg, &curmsgs, list) {
3944 			if (!strcmp(msg->id, args[1])) {
3945 				ha_alert("parsing [%s:%d]: spoe-message section '%s' has the same"
3946 					 " name as another one declared at %s:%d.\n",
3947 					 file, linenum, args[1], msg->conf.file, msg->conf.line);
3948 				err_code |= ERR_ALERT | ERR_FATAL;
3949 				goto out;
3950 			}
3951 		}
3952 
3953 		if ((curmsg = calloc(1, sizeof(*curmsg))) == NULL) {
3954 			ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3955 			err_code |= ERR_ALERT | ERR_ABORT;
3956 			goto out;
3957 		}
3958 
3959 		curmsg->id = strdup(args[1]);
3960 		curmsg->id_len = strlen(curmsg->id);
3961 		curmsg->event  = SPOE_EV_NONE;
3962 		curmsg->conf.file = strdup(file);
3963 		curmsg->conf.line = linenum;
3964 		curmsg->nargs = 0;
3965 		LIST_INIT(&curmsg->args);
3966 		LIST_INIT(&curmsg->acls);
3967 		LIST_INIT(&curmsg->by_evt);
3968 		LIST_INIT(&curmsg->by_grp);
3969 		LIST_ADDQ(&curmsgs, &curmsg->list);
3970 	}
3971 	else if (!strcmp(args[0], "args")) {
3972 		int cur_arg = 1;
3973 
3974 		curproxy->conf.args.ctx  = ARGC_SPOE;
3975 		curproxy->conf.args.file = file;
3976 		curproxy->conf.args.line = linenum;
3977 		while (*args[cur_arg]) {
3978 			char *delim = strchr(args[cur_arg], '=');
3979 			int   idx = 0;
3980 
3981 			if ((arg = calloc(1, sizeof(*arg))) == NULL) {
3982 				ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
3983 				err_code |= ERR_ALERT | ERR_ABORT;
3984 				goto out;
3985 			}
3986 
3987 			if (!delim) {
3988 				arg->name = NULL;
3989 				arg->name_len  = 0;
3990 				delim = args[cur_arg];
3991 			}
3992 			else {
3993 				arg->name = my_strndup(args[cur_arg], delim - args[cur_arg]);
3994 				arg->name_len = delim - args[cur_arg];
3995 				delim++;
3996 			}
3997 			arg->expr = sample_parse_expr((char*[]){delim, NULL},
3998 						      &idx, file, linenum, &errmsg,
3999 						      &curproxy->conf.args);
4000 			if (arg->expr == NULL) {
4001 				ha_alert("parsing [%s:%d] : '%s': %s.\n", file, linenum, args[0], errmsg);
4002 				err_code |= ERR_ALERT | ERR_FATAL;
4003 				free(arg->name);
4004 				free(arg);
4005 				goto out;
4006 			}
4007 			curmsg->nargs++;
4008 			LIST_ADDQ(&curmsg->args, &arg->list);
4009 			cur_arg++;
4010 		}
4011 		curproxy->conf.args.file = NULL;
4012 		curproxy->conf.args.line = 0;
4013 	}
4014 	else if (!strcmp(args[0], "acl")) {
4015 		err = invalid_char(args[1]);
4016 		if (err) {
4017 			ha_alert("parsing [%s:%d] : character '%c' is not permitted in acl name '%s'.\n",
4018 				 file, linenum, *err, args[1]);
4019 			err_code |= ERR_ALERT | ERR_FATAL;
4020 			goto out;
4021 		}
4022 		if (strcasecmp(args[1], "or") == 0) {
4023 			ha_warning("parsing [%s:%d] : acl name '%s' will never match. 'or' is used to express a "
4024 				   "logical disjunction within a condition.\n",
4025 				   file, linenum, args[1]);
4026 			err_code |= ERR_WARN;
4027 		}
4028 		if (parse_acl((const char **)args + 1, &curmsg->acls, &errmsg, &curproxy->conf.args, file, linenum) == NULL) {
4029 			ha_alert("parsing [%s:%d] : error detected while parsing ACL '%s' : %s.\n",
4030 				 file, linenum, args[1], errmsg);
4031 			err_code |= ERR_ALERT | ERR_FATAL;
4032 			goto out;
4033 		}
4034 	}
4035 	else if (!strcmp(args[0], "event")) {
4036 		if (!*args[1]) {
4037 			ha_alert("parsing [%s:%d] : missing event name.\n", file, linenum);
4038 			err_code |= ERR_ALERT | ERR_FATAL;
4039 			goto out;
4040 		}
4041 		/* if (alertif_too_many_args(1, file, linenum, args, &err_code)) */
4042 		/* 	goto out; */
4043 
4044 		if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_CLIENT_SESS]))
4045 			curmsg->event = SPOE_EV_ON_CLIENT_SESS;
4046 		else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_SERVER_SESS]))
4047 			curmsg->event = SPOE_EV_ON_SERVER_SESS;
4048 
4049 		else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_REQ_FE]))
4050 			curmsg->event = SPOE_EV_ON_TCP_REQ_FE;
4051 		else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_REQ_BE]))
4052 			curmsg->event = SPOE_EV_ON_TCP_REQ_BE;
4053 		else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_TCP_RSP]))
4054 			curmsg->event = SPOE_EV_ON_TCP_RSP;
4055 
4056 		else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_HTTP_REQ_FE]))
4057 			curmsg->event = SPOE_EV_ON_HTTP_REQ_FE;
4058 		else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_HTTP_REQ_BE]))
4059 			curmsg->event = SPOE_EV_ON_HTTP_REQ_BE;
4060 		else if (!strcmp(args[1], spoe_event_str[SPOE_EV_ON_HTTP_RSP]))
4061 			curmsg->event = SPOE_EV_ON_HTTP_RSP;
4062 		else {
4063 			ha_alert("parsing [%s:%d] : unknown event '%s'.\n",
4064 				 file, linenum, args[1]);
4065 			err_code |= ERR_ALERT | ERR_FATAL;
4066 			goto out;
4067 		}
4068 
4069 		if (strcmp(args[2], "if") == 0 || strcmp(args[2], "unless") == 0) {
4070 			struct acl_cond *cond;
4071 
4072 			cond = build_acl_cond(file, linenum, &curmsg->acls,
4073 					      curproxy, (const char **)args+2,
4074 					      &errmsg);
4075 			if (cond == NULL) {
4076 				ha_alert("parsing [%s:%d] : error detected while "
4077 					 "parsing an 'event %s' condition : %s.\n",
4078 					 file, linenum, args[1], errmsg);
4079 				err_code |= ERR_ALERT | ERR_FATAL;
4080 				goto out;
4081 			}
4082 			curmsg->cond = cond;
4083 		}
4084 		else if (*args[2]) {
4085 			ha_alert("parsing [%s:%d]: 'event %s' expects either 'if' "
4086 				 "or 'unless' followed by a condition but found '%s'.\n",
4087 				 file, linenum, args[1], args[2]);
4088 			err_code |= ERR_ALERT | ERR_FATAL;
4089 			goto out;
4090 		}
4091 	}
4092 	else if (!*args[0]) {
4093 		ha_alert("parsing [%s:%d] : unknown keyword '%s' in spoe-message section.\n",
4094 			 file, linenum, args[0]);
4095 		err_code |= ERR_ALERT | ERR_FATAL;
4096 		goto out;
4097 	}
4098  out:
4099 	free(errmsg);
4100 	return err_code;
4101 }
4102 
4103 /* Return -1 on error, else 0 */
4104 static int
parse_spoe_flt(char ** args,int * cur_arg,struct proxy * px,struct flt_conf * fconf,char ** err,void * private)4105 parse_spoe_flt(char **args, int *cur_arg, struct proxy *px,
4106                 struct flt_conf *fconf, char **err, void *private)
4107 {
4108 	struct list backup_sections;
4109 	struct spoe_config          *conf;
4110 	struct spoe_message         *msg, *msgback;
4111 	struct spoe_group           *grp, *grpback;
4112 	struct spoe_placeholder     *ph, *phback;
4113 	struct spoe_var_placeholder *vph, *vphback;
4114 	struct logsrv               *logsrv, *logsrvback;
4115 	char                        *file = NULL, *engine = NULL;
4116 	int                          ret, pos = *cur_arg + 1;
4117 
4118 	LIST_INIT(&curmsgs);
4119 	LIST_INIT(&curgrps);
4120 	LIST_INIT(&curmphs);
4121 	LIST_INIT(&curgphs);
4122 	LIST_INIT(&curvars);
4123 	LIST_INIT(&curlogsrvs);
4124 	curpxopts  = 0;
4125 	curpxopts2 = 0;
4126 
4127 	conf = calloc(1, sizeof(*conf));
4128 	if (conf == NULL) {
4129 		memprintf(err, "%s: out of memory", args[*cur_arg]);
4130 		goto error;
4131 	}
4132 	conf->proxy = px;
4133 
4134 	while (*args[pos]) {
4135 		if (!strcmp(args[pos], "config")) {
4136 			if (!*args[pos+1]) {
4137 				memprintf(err, "'%s' : '%s' option without value",
4138 					  args[*cur_arg], args[pos]);
4139 				goto error;
4140 			}
4141 			file = args[pos+1];
4142 			pos += 2;
4143 		}
4144 		else if (!strcmp(args[pos], "engine")) {
4145 			if (!*args[pos+1]) {
4146 				memprintf(err, "'%s' : '%s' option without value",
4147 					  args[*cur_arg], args[pos]);
4148 				goto error;
4149 			}
4150 			engine = args[pos+1];
4151 			pos += 2;
4152 		}
4153 		else {
4154 			memprintf(err, "unknown keyword '%s'", args[pos]);
4155 			goto error;
4156 		}
4157 	}
4158 	if (file == NULL) {
4159 		memprintf(err, "'%s' : missing config file", args[*cur_arg]);
4160 		goto error;
4161 	}
4162 
4163 	/* backup sections and register SPOE sections */
4164 	LIST_INIT(&backup_sections);
4165 	cfg_backup_sections(&backup_sections);
4166 	cfg_register_section("spoe-agent",   cfg_parse_spoe_agent, NULL);
4167 	cfg_register_section("spoe-group",   cfg_parse_spoe_group, NULL);
4168 	cfg_register_section("spoe-message", cfg_parse_spoe_message, NULL);
4169 
4170 	/* Parse SPOE filter configuration file */
4171 	curengine = engine;
4172 	curproxy  = px;
4173 	curagent  = NULL;
4174 	curmsg    = NULL;
4175 	ret = readcfgfile(file);
4176 	curproxy = NULL;
4177 
4178 	/* unregister SPOE sections and restore previous sections */
4179 	cfg_unregister_sections();
4180 	cfg_restore_sections(&backup_sections);
4181 
4182 	if (ret == -1) {
4183 		memprintf(err, "Could not open configuration file %s : %s",
4184 			  file, strerror(errno));
4185 		goto error;
4186 	}
4187 	if (ret & (ERR_ABORT|ERR_FATAL)) {
4188 		memprintf(err, "Error(s) found in configuration file %s", file);
4189 		goto error;
4190 	}
4191 
4192 	/* Check SPOE agent */
4193 	if (curagent == NULL) {
4194 		memprintf(err, "No SPOE agent found in file %s", file);
4195 		goto error;
4196 	}
4197 	if (curagent->b.name == NULL) {
4198 		memprintf(err, "No backend declared for SPOE agent '%s' declared at %s:%d",
4199 			  curagent->id, curagent->conf.file, curagent->conf.line);
4200 		goto error;
4201 	}
4202 	if (curagent->timeout.hello      == TICK_ETERNITY ||
4203 	    curagent->timeout.idle       == TICK_ETERNITY ||
4204 	    curagent->timeout.processing == TICK_ETERNITY) {
4205 		ha_warning("Proxy '%s': missing timeouts for SPOE agent '%s' declare at %s:%d.\n"
4206 			   "   | While not properly invalid, you will certainly encounter various problems\n"
4207 			   "   | with such a configuration. To fix this, please ensure that all following\n"
4208 			   "   | timeouts are set to a non-zero value: 'hello', 'idle', 'processing'.\n",
4209 			   px->id, curagent->id, curagent->conf.file, curagent->conf.line);
4210 	}
4211 	if (curagent->var_pfx == NULL) {
4212 		char *tmp = curagent->id;
4213 
4214 		while (*tmp) {
4215 			if (!isalnum(*tmp) && *tmp != '_' && *tmp != '.') {
4216 				memprintf(err, "Invalid variable prefix '%s' for SPOE agent '%s' declared at %s:%d. "
4217 					  "Use 'option var-prefix' to set it. Only [a-zA-Z0-9_.] chars are supported.\n",
4218 					  curagent->id, curagent->id, curagent->conf.file, curagent->conf.line);
4219 				goto error;
4220 			}
4221 			tmp++;
4222 		}
4223 		curagent->var_pfx = strdup(curagent->id);
4224 	}
4225 
4226 	if (curagent->var_on_error) {
4227 		struct arg arg;
4228 
4229 		trash.data = snprintf(trash.area, trash.size, "txn.%s.%s",
4230 				     curagent->var_pfx, curagent->var_on_error);
4231 
4232 		arg.type = ARGT_STR;
4233 		arg.data.str.area = trash.area;
4234 		arg.data.str.data = trash.data;
4235 		arg.data.str.size = 0; /* Set it to 0 to not release it in vars_check_args() */
4236 		if (!vars_check_arg(&arg, err)) {
4237 			memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4238 				  curagent->id, curagent->var_pfx, curagent->var_on_error, *err);
4239 			goto error;
4240 		}
4241 	}
4242 
4243 	if (curagent->var_t_process) {
4244 		struct arg arg;
4245 
4246 		trash.data = snprintf(trash.area, trash.size, "txn.%s.%s",
4247 				     curagent->var_pfx, curagent->var_t_process);
4248 
4249 		arg.type = ARGT_STR;
4250 		arg.data.str.area = trash.area;
4251 		arg.data.str.data = trash.data;
4252 		arg.data.str.size = 0;  /* Set it to 0 to not release it in vars_check_args() */
4253 		if (!vars_check_arg(&arg, err)) {
4254 			memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4255 				  curagent->id, curagent->var_pfx, curagent->var_t_process, *err);
4256 			goto error;
4257 		}
4258 	}
4259 
4260 	if (curagent->var_t_total) {
4261 		struct arg arg;
4262 
4263 		trash.data = snprintf(trash.area, trash.size, "txn.%s.%s",
4264 				     curagent->var_pfx, curagent->var_t_total);
4265 
4266 		arg.type = ARGT_STR;
4267 		arg.data.str.area = trash.area;
4268 		arg.data.str.data = trash.data;
4269 		arg.data.str.size = 0;  /* Set it to 0 to not release it in vars_check_args() */
4270 		if (!vars_check_arg(&arg, err)) {
4271 			memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4272 				  curagent->id, curagent->var_pfx, curagent->var_t_process, *err);
4273 			goto error;
4274 		}
4275 	}
4276 
4277 	if (LIST_ISEMPTY(&curmphs) && LIST_ISEMPTY(&curgphs)) {
4278 		ha_warning("Proxy '%s': No message/group used by SPOE agent '%s' declared at %s:%d.\n",
4279 			   px->id, curagent->id, curagent->conf.file, curagent->conf.line);
4280 		goto finish;
4281 	}
4282 
4283 	/* Replace placeholders by the corresponding messages for the SPOE
4284 	 * agent */
4285 	list_for_each_entry(ph, &curmphs, list) {
4286 		list_for_each_entry(msg, &curmsgs, list) {
4287 			struct spoe_arg *arg;
4288 			unsigned int     where;
4289 
4290 			if (!strcmp(msg->id, ph->id)) {
4291 				if ((px->cap & (PR_CAP_FE|PR_CAP_BE)) == (PR_CAP_FE|PR_CAP_BE)) {
4292 					if (msg->event == SPOE_EV_ON_TCP_REQ_BE)
4293 						msg->event = SPOE_EV_ON_TCP_REQ_FE;
4294 					if (msg->event == SPOE_EV_ON_HTTP_REQ_BE)
4295 						msg->event = SPOE_EV_ON_HTTP_REQ_FE;
4296 				}
4297 				if (!(px->cap & PR_CAP_FE) && (msg->event == SPOE_EV_ON_CLIENT_SESS ||
4298 							       msg->event == SPOE_EV_ON_TCP_REQ_FE ||
4299 							       msg->event == SPOE_EV_ON_HTTP_REQ_FE)) {
4300 					ha_warning("Proxy '%s': frontend event used on a backend proxy at %s:%d.\n",
4301 						   px->id, msg->conf.file, msg->conf.line);
4302 					goto next_mph;
4303 				}
4304 				if (msg->event == SPOE_EV_NONE) {
4305 					ha_warning("Proxy '%s': Ignore SPOE message '%s' without event at %s:%d.\n",
4306 						   px->id, msg->id, msg->conf.file, msg->conf.line);
4307 					goto next_mph;
4308 				}
4309 
4310 				where = 0;
4311 				switch (msg->event) {
4312 					case SPOE_EV_ON_CLIENT_SESS:
4313 						where |= SMP_VAL_FE_CON_ACC;
4314 						break;
4315 
4316 					case SPOE_EV_ON_TCP_REQ_FE:
4317 						where |= SMP_VAL_FE_REQ_CNT;
4318 						break;
4319 
4320 					case SPOE_EV_ON_HTTP_REQ_FE:
4321 						where |= SMP_VAL_FE_HRQ_HDR;
4322 						break;
4323 
4324 					case SPOE_EV_ON_TCP_REQ_BE:
4325 						if (px->cap & PR_CAP_FE)
4326 							where |= SMP_VAL_FE_REQ_CNT;
4327 						if (px->cap & PR_CAP_BE)
4328 							where |= SMP_VAL_BE_REQ_CNT;
4329 						break;
4330 
4331 					case SPOE_EV_ON_HTTP_REQ_BE:
4332 						if (px->cap & PR_CAP_FE)
4333 							where |= SMP_VAL_FE_HRQ_HDR;
4334 						if (px->cap & PR_CAP_BE)
4335 							where |= SMP_VAL_BE_HRQ_HDR;
4336 						break;
4337 
4338 					case SPOE_EV_ON_SERVER_SESS:
4339 						where |= SMP_VAL_BE_SRV_CON;
4340 						break;
4341 
4342 					case SPOE_EV_ON_TCP_RSP:
4343 						if (px->cap & PR_CAP_FE)
4344 							where |= SMP_VAL_FE_RES_CNT;
4345 						if (px->cap & PR_CAP_BE)
4346 							where |= SMP_VAL_BE_RES_CNT;
4347 						break;
4348 
4349 					case SPOE_EV_ON_HTTP_RSP:
4350 						if (px->cap & PR_CAP_FE)
4351 							where |= SMP_VAL_FE_HRS_HDR;
4352 						if (px->cap & PR_CAP_BE)
4353 							where |= SMP_VAL_BE_HRS_HDR;
4354 						break;
4355 
4356 					default:
4357 						break;
4358 				}
4359 
4360 				list_for_each_entry(arg, &msg->args, list) {
4361 					if (!(arg->expr->fetch->val & where)) {
4362 						memprintf(err, "Ignore SPOE message '%s' at %s:%d: "
4363 							"some args extract information from '%s', "
4364 							"none of which is available here ('%s')",
4365 							msg->id, msg->conf.file, msg->conf.line,
4366 							sample_ckp_names(arg->expr->fetch->use),
4367 							sample_ckp_names(where));
4368 						goto error;
4369 					}
4370 				}
4371 
4372 				msg->agent = curagent;
4373 				LIST_ADDQ(&curagent->events[msg->event], &msg->by_evt);
4374 				goto next_mph;
4375 			}
4376 		}
4377 		memprintf(err, "SPOE agent '%s' try to use undefined SPOE message '%s' at %s:%d",
4378 			  curagent->id, ph->id, curagent->conf.file, curagent->conf.line);
4379 		goto error;
4380 	  next_mph:
4381 		continue;
4382 	}
4383 
4384 	/* Replace placeholders by the corresponding groups for the SPOE
4385 	 * agent */
4386 	list_for_each_entry(ph, &curgphs, list) {
4387 		list_for_each_entry_safe(grp, grpback, &curgrps, list) {
4388 			if (!strcmp(grp->id, ph->id)) {
4389 				grp->agent = curagent;
4390 				LIST_DEL(&grp->list);
4391 				LIST_ADDQ(&curagent->groups, &grp->list);
4392 				goto next_aph;
4393 			}
4394 		}
4395 		memprintf(err, "SPOE agent '%s' try to use undefined SPOE group '%s' at %s:%d",
4396 			  curagent->id, ph->id, curagent->conf.file, curagent->conf.line);
4397 		goto error;
4398 	  next_aph:
4399 		continue;
4400 	}
4401 
4402 	/* Replace placeholders by the corresponding message for each SPOE
4403 	 * group of the SPOE agent */
4404 	list_for_each_entry(grp, &curagent->groups, list) {
4405 		list_for_each_entry_safe(ph, phback, &grp->phs, list) {
4406 			list_for_each_entry(msg, &curmsgs, list) {
4407 				if (!strcmp(msg->id, ph->id)) {
4408 					if (msg->group != NULL) {
4409 						memprintf(err, "SPOE message '%s' already belongs to "
4410 							  "the SPOE group '%s' declare at %s:%d",
4411 							  msg->id, msg->group->id,
4412 							  msg->group->conf.file,
4413 							  msg->group->conf.line);
4414 						goto error;
4415 					}
4416 
4417 					/* Scope for arguments are not checked for now. We will check
4418 					 * them only if a rule use the corresponding SPOE group. */
4419 					msg->agent = curagent;
4420 					msg->group = grp;
4421 					LIST_DEL(&ph->list);
4422 					LIST_ADDQ(&grp->messages, &msg->by_grp);
4423 					goto next_mph_grp;
4424 				}
4425 			}
4426 			memprintf(err, "SPOE group '%s' try to use undefined SPOE message '%s' at %s:%d",
4427 				  grp->id, ph->id, curagent->conf.file, curagent->conf.line);
4428 			goto error;
4429 		  next_mph_grp:
4430 			continue;
4431 		}
4432 	}
4433 
4434  finish:
4435 	/* move curmsgs to the agent message list */
4436 	curmsgs.n->p = &curagent->messages;
4437 	curmsgs.p->n = &curagent->messages;
4438 	curagent->messages = curmsgs;
4439 	LIST_INIT(&curmsgs);
4440 
4441 	conf->id    = strdup(engine ? engine : curagent->id);
4442 	conf->agent = curagent;
4443 
4444 	/* Start agent's proxy initialization here. It will be finished during
4445 	 * the filter init. */
4446         memset(&conf->agent_fe, 0, sizeof(conf->agent_fe));
4447         init_new_proxy(&conf->agent_fe);
4448 	conf->agent_fe.id        = conf->agent->id;
4449 	conf->agent_fe.parent    = conf->agent;
4450 	conf->agent_fe.options  |= curpxopts;
4451 	conf->agent_fe.options2 |= curpxopts2;
4452 
4453 	list_for_each_entry_safe(logsrv, logsrvback, &curlogsrvs, list) {
4454 		LIST_DEL(&logsrv->list);
4455 		LIST_ADDQ(&conf->agent_fe.logsrvs, &logsrv->list);
4456 	}
4457 
4458 	list_for_each_entry_safe(ph, phback, &curmphs, list) {
4459 		LIST_DEL(&ph->list);
4460 		spoe_release_placeholder(ph);
4461 	}
4462 	list_for_each_entry_safe(ph, phback, &curgphs, list) {
4463 		LIST_DEL(&ph->list);
4464 		spoe_release_placeholder(ph);
4465 	}
4466 	list_for_each_entry_safe(vph, vphback, &curvars, list) {
4467 		struct arg arg;
4468 
4469 		trash.data = snprintf(trash.area, trash.size, "proc.%s.%s",
4470 				     curagent->var_pfx, vph->name);
4471 
4472 		arg.type = ARGT_STR;
4473 		arg.data.str.area = trash.area;
4474 		arg.data.str.data = trash.data;
4475 		arg.data.str.size = 0;  /* Set it to 0 to not release it in vars_check_args() */
4476 		if (!vars_check_arg(&arg, err)) {
4477 			memprintf(err, "SPOE agent '%s': failed to register variable %s.%s (%s)",
4478 				  curagent->id, curagent->var_pfx, vph->name, *err);
4479 			goto error;
4480 		}
4481 
4482 		LIST_DEL(&vph->list);
4483 		free(vph->name);
4484 		free(vph);
4485 	}
4486 	list_for_each_entry_safe(grp, grpback, &curgrps, list) {
4487 		LIST_DEL(&grp->list);
4488 		spoe_release_group(grp);
4489 	}
4490 	*cur_arg    = pos;
4491 	fconf->id   = spoe_filter_id;
4492 	fconf->ops  = &spoe_ops;
4493 	fconf->conf = conf;
4494 	return 0;
4495 
4496  error:
4497 	spoe_release_agent(curagent);
4498 	list_for_each_entry_safe(ph, phback, &curmphs, list) {
4499 		LIST_DEL(&ph->list);
4500 		spoe_release_placeholder(ph);
4501 	}
4502 	list_for_each_entry_safe(ph, phback, &curgphs, list) {
4503 		LIST_DEL(&ph->list);
4504 		spoe_release_placeholder(ph);
4505 	}
4506 	list_for_each_entry_safe(vph, vphback, &curvars, list) {
4507 		LIST_DEL(&vph->list);
4508 		free(vph->name);
4509 		free(vph);
4510 	}
4511 	list_for_each_entry_safe(grp, grpback, &curgrps, list) {
4512 		LIST_DEL(&grp->list);
4513 		spoe_release_group(grp);
4514 	}
4515 	list_for_each_entry_safe(msg, msgback, &curmsgs, list) {
4516 		LIST_DEL(&msg->list);
4517 		spoe_release_message(msg);
4518 	}
4519 	list_for_each_entry_safe(logsrv, logsrvback, &curlogsrvs, list) {
4520 		LIST_DEL(&logsrv->list);
4521 		free(logsrv);
4522 	}
4523 	free(conf);
4524 	return -1;
4525 }
4526 
4527 /* Send message of a SPOE group. This is the action_ptr callback of a rule
4528  * associated to a "send-spoe-group" action.
4529  *
4530  * It returns ACT_RET_CONT is processing is finished without error, it returns
4531  * ACT_RET_YIELD if the action is in progress. Otherwise it returns
4532  * ACT_RET_ERR. */
4533 static enum act_return
spoe_send_group(struct act_rule * rule,struct proxy * px,struct session * sess,struct stream * s,int flags)4534 spoe_send_group(struct act_rule *rule, struct proxy *px,
4535 		struct session *sess, struct stream *s, int flags)
4536 {
4537 	struct filter      *filter;
4538 	struct spoe_agent   *agent = NULL;
4539 	struct spoe_group   *group = NULL;
4540 	struct spoe_context *ctx   = NULL;
4541 	int ret, dir;
4542 
4543 	list_for_each_entry(filter, &s->strm_flt.filters, list) {
4544 		if (filter->config == rule->arg.act.p[0]) {
4545 			agent = rule->arg.act.p[2];
4546 			group = rule->arg.act.p[3];
4547 			ctx   = filter->ctx;
4548 			break;
4549 		}
4550 	}
4551 	if (agent == NULL || group == NULL || ctx == NULL)
4552 		return ACT_RET_ERR;
4553 	if (ctx->state == SPOE_CTX_ST_NONE)
4554 		return ACT_RET_CONT;
4555 
4556 	switch (rule->from) {
4557 		case ACT_F_TCP_REQ_SES: dir = SMP_OPT_DIR_REQ; break;
4558 		case ACT_F_TCP_REQ_CNT: dir = SMP_OPT_DIR_REQ; break;
4559 		case ACT_F_TCP_RES_CNT: dir = SMP_OPT_DIR_RES; break;
4560 		case ACT_F_HTTP_REQ:    dir = SMP_OPT_DIR_REQ; break;
4561 		case ACT_F_HTTP_RES:    dir = SMP_OPT_DIR_RES; break;
4562 		default:
4563 			SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
4564 				    " - internal error while execute spoe-send-group\n",
4565 				    (int)now.tv_sec, (int)now.tv_usec, agent->id,
4566 				    __FUNCTION__, s);
4567 			send_log(px, LOG_ERR, "SPOE: [%s] internal error while execute spoe-send-group\n",
4568 				 agent->id);
4569 			return ACT_RET_CONT;
4570 	}
4571 
4572 	ret = spoe_process_group(s, ctx, group, dir);
4573 	if (ret == 1)
4574 		return ACT_RET_CONT;
4575 	else if (ret == 0) {
4576 		if (flags & ACT_FLAG_FINAL) {
4577 			SPOE_PRINTF(stderr, "%d.%06d [SPOE/%-15s] %s: stream=%p"
4578 				    " - failed to process group '%s': interrupted by caller\n",
4579 				    (int)now.tv_sec, (int)now.tv_usec,
4580 				    agent->id, __FUNCTION__, s, group->id);
4581 			ctx->status_code = SPOE_CTX_ERR_INTERRUPT;
4582 			spoe_stop_processing(agent, ctx);
4583 			spoe_handle_processing_error(s, agent, ctx, dir);
4584 			return ACT_RET_CONT;
4585 		}
4586 		return ACT_RET_YIELD;
4587 	}
4588 	else
4589 		return ACT_RET_ERR;
4590 }
4591 
4592 /* Check an "send-spoe-group" action. Here, we'll try to find the real SPOE
4593  * group associated to <rule>. The format of an rule using 'send-spoe-group'
4594  * action should be:
4595  *
4596  *   (http|tcp)-(request|response) send-spoe-group <engine-id> <group-id>
4597  *
4598  * So, we'll loop on each configured SPOE filter for the proxy <px> to find the
4599  * SPOE engine matching <engine-id>. And then, we'll try to find the good group
4600  * matching <group-id>. Finally, we'll check all messages referenced by the SPOE
4601  * group.
4602  *
4603  * The function returns 1 in success case, otherwise, it returns 0 and err is
4604  * filled.
4605  */
4606 static int
check_send_spoe_group(struct act_rule * rule,struct proxy * px,char ** err)4607 check_send_spoe_group(struct act_rule *rule, struct proxy *px, char **err)
4608 {
4609 	struct flt_conf     *fconf;
4610 	struct spoe_config  *conf;
4611 	struct spoe_agent   *agent = NULL;
4612 	struct spoe_group   *group;
4613 	struct spoe_message *msg;
4614 	char                *engine_id = rule->arg.act.p[0];
4615 	char                *group_id  = rule->arg.act.p[1];
4616 	unsigned int         where = 0;
4617 
4618 	switch (rule->from) {
4619 		case ACT_F_TCP_REQ_SES: where = SMP_VAL_FE_SES_ACC; break;
4620 		case ACT_F_TCP_REQ_CNT: where = SMP_VAL_FE_REQ_CNT; break;
4621 		case ACT_F_TCP_RES_CNT: where = SMP_VAL_BE_RES_CNT; break;
4622 		case ACT_F_HTTP_REQ:    where = SMP_VAL_FE_HRQ_HDR; break;
4623 		case ACT_F_HTTP_RES:    where = SMP_VAL_BE_HRS_HDR; break;
4624 		default:
4625 			memprintf(err,
4626 				  "internal error, unexpected rule->from=%d, please report this bug!",
4627 				  rule->from);
4628 			goto error;
4629 	}
4630 
4631 	/* Try to find the SPOE engine by checking all SPOE filters for proxy
4632 	 * <px> */
4633 	list_for_each_entry(fconf, &px->filter_configs, list) {
4634 		conf = fconf->conf;
4635 
4636 		/* This is not an SPOE filter */
4637 		if (fconf->id != spoe_filter_id)
4638 			continue;
4639 
4640 		/* This is the good engine */
4641 		if (!strcmp(conf->id, engine_id)) {
4642 			agent = conf->agent;
4643 			break;
4644 		}
4645 	}
4646 	if (agent == NULL) {
4647 		memprintf(err, "unable to find SPOE engine '%s' used by the send-spoe-group '%s'",
4648 			  engine_id, group_id);
4649 		goto error;
4650 	}
4651 
4652 	/* Try to find the right group */
4653 	list_for_each_entry(group, &agent->groups, list) {
4654 		/* This is the good group */
4655 		if (!strcmp(group->id, group_id))
4656 			break;
4657 	}
4658 	if (&group->list == &agent->groups) {
4659 		memprintf(err, "unable to find SPOE group '%s' into SPOE engine '%s' configuration",
4660 			  group_id, engine_id);
4661 		goto error;
4662 	}
4663 
4664 	/* Ok, we found the group, we need to check messages and their
4665 	 * arguments */
4666 	list_for_each_entry(msg, &group->messages, by_grp) {
4667 		struct spoe_arg *arg;
4668 
4669 		list_for_each_entry(arg, &msg->args, list) {
4670 			if (!(arg->expr->fetch->val & where)) {
4671 				memprintf(err, "Invalid SPOE message '%s' used by SPOE group '%s' at %s:%d: "
4672 					  "some args extract information from '%s',"
4673 					  "none of which is available here ('%s')",
4674 					  msg->id, group->id, msg->conf.file, msg->conf.line,
4675 					  sample_ckp_names(arg->expr->fetch->use),
4676 					  sample_ckp_names(where));
4677 				goto error;
4678 			}
4679 		}
4680 	}
4681 
4682 	free(engine_id);
4683 	free(group_id);
4684 	rule->arg.act.p[0] = fconf; /* Associate filter config with the rule */
4685 	rule->arg.act.p[1] = conf;  /* Associate SPOE config with the rule */
4686 	rule->arg.act.p[2] = agent; /* Associate SPOE agent with the rule */
4687 	rule->arg.act.p[3] = group; /* Associate SPOE group with the rule */
4688 	return 1;
4689 
4690   error:
4691 	free(engine_id);
4692 	free(group_id);
4693 	return 0;
4694 }
4695 
4696 /* Parse 'send-spoe-group' action following the format:
4697  *
4698  *     ... send-spoe-group <engine-id> <group-id>
4699  *
4700  * It returns ACT_RET_PRS_ERR if fails and <err> is filled with an error
4701  * message. Otherwise, it returns ACT_RET_PRS_OK and parsing engine and group
4702  * ids are saved and used later, when the rule will be checked.
4703  */
4704 static enum act_parse_ret
parse_send_spoe_group(const char ** args,int * orig_arg,struct proxy * px,struct act_rule * rule,char ** err)4705 parse_send_spoe_group(const char **args, int *orig_arg, struct proxy *px,
4706 		      struct act_rule *rule, char **err)
4707 {
4708 	if (!*args[*orig_arg] || !*args[*orig_arg+1] ||
4709 	    (*args[*orig_arg+2] && strcmp(args[*orig_arg+2], "if") != 0 && strcmp(args[*orig_arg+2], "unless") != 0)) {
4710 		memprintf(err, "expects 2 arguments: <engine-id> <group-id>");
4711 		return ACT_RET_PRS_ERR;
4712 	}
4713 	rule->arg.act.p[0] = strdup(args[*orig_arg]);   /* Copy the SPOE engine id */
4714 	rule->arg.act.p[1] = strdup(args[*orig_arg+1]); /* Cope the SPOE group id */
4715 
4716 	(*orig_arg) += 2;
4717 
4718 	rule->action     = ACT_CUSTOM;
4719 	rule->action_ptr = spoe_send_group;
4720 	rule->check_ptr  = check_send_spoe_group;
4721 	return ACT_RET_PRS_OK;
4722 }
4723 
4724 
4725 /* Declare the filter parser for "spoe" keyword */
4726 static struct flt_kw_list flt_kws = { "SPOE", { }, {
4727 		{ "spoe", parse_spoe_flt, NULL },
4728 		{ NULL, NULL, NULL },
4729 	}
4730 };
4731 
4732 INITCALL1(STG_REGISTER, flt_register_keywords, &flt_kws);
4733 
4734 /* Delcate the action parser for "spoe-action" keyword */
4735 static struct action_kw_list tcp_req_action_kws = { { }, {
4736 		{ "send-spoe-group", parse_send_spoe_group },
4737 		{ /* END */ },
4738 	}
4739 };
4740 
4741 INITCALL1(STG_REGISTER, tcp_req_cont_keywords_register, &tcp_req_action_kws);
4742 
4743 static struct action_kw_list tcp_res_action_kws = { { }, {
4744 		{ "send-spoe-group", parse_send_spoe_group },
4745 		{ /* END */ },
4746 	}
4747 };
4748 
4749 INITCALL1(STG_REGISTER, tcp_res_cont_keywords_register, &tcp_res_action_kws);
4750 
4751 static struct action_kw_list http_req_action_kws = { { }, {
4752 		{ "send-spoe-group", parse_send_spoe_group },
4753 		{ /* END */ },
4754 	}
4755 };
4756 
4757 INITCALL1(STG_REGISTER, http_req_keywords_register, &http_req_action_kws);
4758 
4759 static struct action_kw_list http_res_action_kws = { { }, {
4760 		{ "send-spoe-group", parse_send_spoe_group },
4761 		{ /* END */ },
4762 	}
4763 };
4764 
4765 INITCALL1(STG_REGISTER, http_res_keywords_register, &http_res_action_kws);
4766