strparser: Call sock_owned_by_user_nocheck
[muen/linux.git] / net / strparser / strparser.c
1 /*
2  * Stream Parser
3  *
4  * Copyright (c) 2016 Tom Herbert <tom@herbertland.com>
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2
8  * as published by the Free Software Foundation.
9  */
10
11 #include <linux/bpf.h>
12 #include <linux/errno.h>
13 #include <linux/errqueue.h>
14 #include <linux/file.h>
15 #include <linux/in.h>
16 #include <linux/kernel.h>
17 #include <linux/module.h>
18 #include <linux/net.h>
19 #include <linux/netdevice.h>
20 #include <linux/poll.h>
21 #include <linux/rculist.h>
22 #include <linux/skbuff.h>
23 #include <linux/socket.h>
24 #include <linux/uaccess.h>
25 #include <linux/workqueue.h>
26 #include <net/strparser.h>
27 #include <net/netns/generic.h>
28 #include <net/sock.h>
29
30 static struct workqueue_struct *strp_wq;
31
32 struct _strp_msg {
33         /* Internal cb structure. struct strp_msg must be first for passing
34          * to upper layer.
35          */
36         struct strp_msg strp;
37         int accum_len;
38         int early_eaten;
39 };
40
41 static inline struct _strp_msg *_strp_msg(struct sk_buff *skb)
42 {
43         return (struct _strp_msg *)((void *)skb->cb +
44                 offsetof(struct qdisc_skb_cb, data));
45 }
46
47 /* Lower lock held */
48 static void strp_abort_strp(struct strparser *strp, int err)
49 {
50         /* Unrecoverable error in receive */
51
52         cancel_delayed_work(&strp->msg_timer_work);
53
54         if (strp->stopped)
55                 return;
56
57         strp->stopped = 1;
58
59         if (strp->sk) {
60                 struct sock *sk = strp->sk;
61
62                 /* Report an error on the lower socket */
63                 sk->sk_err = err;
64                 sk->sk_error_report(sk);
65         }
66 }
67
68 static void strp_start_timer(struct strparser *strp, long timeo)
69 {
70         if (timeo)
71                 mod_delayed_work(strp_wq, &strp->msg_timer_work, timeo);
72 }
73
74 /* Lower lock held */
75 static void strp_parser_err(struct strparser *strp, int err,
76                             read_descriptor_t *desc)
77 {
78         desc->error = err;
79         kfree_skb(strp->skb_head);
80         strp->skb_head = NULL;
81         strp->cb.abort_parser(strp, err);
82 }
83
84 static inline int strp_peek_len(struct strparser *strp)
85 {
86         if (strp->sk) {
87                 struct socket *sock = strp->sk->sk_socket;
88
89                 return sock->ops->peek_len(sock);
90         }
91
92         /* If we don't have an associated socket there's nothing to peek.
93          * Return int max to avoid stopping the strparser.
94          */
95
96         return INT_MAX;
97 }
98
99 /* Lower socket lock held */
100 static int __strp_recv(read_descriptor_t *desc, struct sk_buff *orig_skb,
101                        unsigned int orig_offset, size_t orig_len,
102                        size_t max_msg_size, long timeo)
103 {
104         struct strparser *strp = (struct strparser *)desc->arg.data;
105         struct _strp_msg *stm;
106         struct sk_buff *head, *skb;
107         size_t eaten = 0, cand_len;
108         ssize_t extra;
109         int err;
110         bool cloned_orig = false;
111
112         if (strp->paused)
113                 return 0;
114
115         head = strp->skb_head;
116         if (head) {
117                 /* Message already in progress */
118
119                 stm = _strp_msg(head);
120                 if (unlikely(stm->early_eaten)) {
121                         /* Already some number of bytes on the receive sock
122                          * data saved in skb_head, just indicate they
123                          * are consumed.
124                          */
125                         eaten = orig_len <= stm->early_eaten ?
126                                 orig_len : stm->early_eaten;
127                         stm->early_eaten -= eaten;
128
129                         return eaten;
130                 }
131
132                 if (unlikely(orig_offset)) {
133                         /* Getting data with a non-zero offset when a message is
134                          * in progress is not expected. If it does happen, we
135                          * need to clone and pull since we can't deal with
136                          * offsets in the skbs for a message expect in the head.
137                          */
138                         orig_skb = skb_clone(orig_skb, GFP_ATOMIC);
139                         if (!orig_skb) {
140                                 STRP_STATS_INCR(strp->stats.mem_fail);
141                                 desc->error = -ENOMEM;
142                                 return 0;
143                         }
144                         if (!pskb_pull(orig_skb, orig_offset)) {
145                                 STRP_STATS_INCR(strp->stats.mem_fail);
146                                 kfree_skb(orig_skb);
147                                 desc->error = -ENOMEM;
148                                 return 0;
149                         }
150                         cloned_orig = true;
151                         orig_offset = 0;
152                 }
153
154                 if (!strp->skb_nextp) {
155                         /* We are going to append to the frags_list of head.
156                          * Need to unshare the frag_list.
157                          */
158                         err = skb_unclone(head, GFP_ATOMIC);
159                         if (err) {
160                                 STRP_STATS_INCR(strp->stats.mem_fail);
161                                 desc->error = err;
162                                 return 0;
163                         }
164
165                         if (unlikely(skb_shinfo(head)->frag_list)) {
166                                 /* We can't append to an sk_buff that already
167                                  * has a frag_list. We create a new head, point
168                                  * the frag_list of that to the old head, and
169                                  * then are able to use the old head->next for
170                                  * appending to the message.
171                                  */
172                                 if (WARN_ON(head->next)) {
173                                         desc->error = -EINVAL;
174                                         return 0;
175                                 }
176
177                                 skb = alloc_skb(0, GFP_ATOMIC);
178                                 if (!skb) {
179                                         STRP_STATS_INCR(strp->stats.mem_fail);
180                                         desc->error = -ENOMEM;
181                                         return 0;
182                                 }
183                                 skb->len = head->len;
184                                 skb->data_len = head->len;
185                                 skb->truesize = head->truesize;
186                                 *_strp_msg(skb) = *_strp_msg(head);
187                                 strp->skb_nextp = &head->next;
188                                 skb_shinfo(skb)->frag_list = head;
189                                 strp->skb_head = skb;
190                                 head = skb;
191                         } else {
192                                 strp->skb_nextp =
193                                     &skb_shinfo(head)->frag_list;
194                         }
195                 }
196         }
197
198         while (eaten < orig_len) {
199                 /* Always clone since we will consume something */
200                 skb = skb_clone(orig_skb, GFP_ATOMIC);
201                 if (!skb) {
202                         STRP_STATS_INCR(strp->stats.mem_fail);
203                         desc->error = -ENOMEM;
204                         break;
205                 }
206
207                 cand_len = orig_len - eaten;
208
209                 head = strp->skb_head;
210                 if (!head) {
211                         head = skb;
212                         strp->skb_head = head;
213                         /* Will set skb_nextp on next packet if needed */
214                         strp->skb_nextp = NULL;
215                         stm = _strp_msg(head);
216                         memset(stm, 0, sizeof(*stm));
217                         stm->strp.offset = orig_offset + eaten;
218                 } else {
219                         /* Unclone since we may be appending to an skb that we
220                          * already share a frag_list with.
221                          */
222                         err = skb_unclone(skb, GFP_ATOMIC);
223                         if (err) {
224                                 STRP_STATS_INCR(strp->stats.mem_fail);
225                                 desc->error = err;
226                                 break;
227                         }
228
229                         stm = _strp_msg(head);
230                         *strp->skb_nextp = skb;
231                         strp->skb_nextp = &skb->next;
232                         head->data_len += skb->len;
233                         head->len += skb->len;
234                         head->truesize += skb->truesize;
235                 }
236
237                 if (!stm->strp.full_len) {
238                         ssize_t len;
239
240                         len = (*strp->cb.parse_msg)(strp, head);
241
242                         if (!len) {
243                                 /* Need more header to determine length */
244                                 if (!stm->accum_len) {
245                                         /* Start RX timer for new message */
246                                         strp_start_timer(strp, timeo);
247                                 }
248                                 stm->accum_len += cand_len;
249                                 eaten += cand_len;
250                                 STRP_STATS_INCR(strp->stats.need_more_hdr);
251                                 WARN_ON(eaten != orig_len);
252                                 break;
253                         } else if (len < 0) {
254                                 if (len == -ESTRPIPE && stm->accum_len) {
255                                         len = -ENODATA;
256                                         strp->unrecov_intr = 1;
257                                 } else {
258                                         strp->interrupted = 1;
259                                 }
260                                 strp_parser_err(strp, len, desc);
261                                 break;
262                         } else if (len > max_msg_size) {
263                                 /* Message length exceeds maximum allowed */
264                                 STRP_STATS_INCR(strp->stats.msg_too_big);
265                                 strp_parser_err(strp, -EMSGSIZE, desc);
266                                 break;
267                         } else if (len <= (ssize_t)head->len -
268                                           skb->len - stm->strp.offset) {
269                                 /* Length must be into new skb (and also
270                                  * greater than zero)
271                                  */
272                                 STRP_STATS_INCR(strp->stats.bad_hdr_len);
273                                 strp_parser_err(strp, -EPROTO, desc);
274                                 break;
275                         }
276
277                         stm->strp.full_len = len;
278                 }
279
280                 extra = (ssize_t)(stm->accum_len + cand_len) -
281                         stm->strp.full_len;
282
283                 if (extra < 0) {
284                         /* Message not complete yet. */
285                         if (stm->strp.full_len - stm->accum_len >
286                             strp_peek_len(strp)) {
287                                 /* Don't have the whole message in the socket
288                                  * buffer. Set strp->need_bytes to wait for
289                                  * the rest of the message. Also, set "early
290                                  * eaten" since we've already buffered the skb
291                                  * but don't consume yet per strp_read_sock.
292                                  */
293
294                                 if (!stm->accum_len) {
295                                         /* Start RX timer for new message */
296                                         strp_start_timer(strp, timeo);
297                                 }
298
299                                 strp->need_bytes = stm->strp.full_len -
300                                                        stm->accum_len;
301                                 stm->accum_len += cand_len;
302                                 stm->early_eaten = cand_len;
303                                 STRP_STATS_ADD(strp->stats.bytes, cand_len);
304                                 desc->count = 0; /* Stop reading socket */
305                                 break;
306                         }
307                         stm->accum_len += cand_len;
308                         eaten += cand_len;
309                         WARN_ON(eaten != orig_len);
310                         break;
311                 }
312
313                 /* Positive extra indicates ore bytes than needed for the
314                  * message
315                  */
316
317                 WARN_ON(extra > cand_len);
318
319                 eaten += (cand_len - extra);
320
321                 /* Hurray, we have a new message! */
322                 cancel_delayed_work(&strp->msg_timer_work);
323                 strp->skb_head = NULL;
324                 STRP_STATS_INCR(strp->stats.msgs);
325
326                 /* Give skb to upper layer */
327                 strp->cb.rcv_msg(strp, head);
328
329                 if (unlikely(strp->paused)) {
330                         /* Upper layer paused strp */
331                         break;
332                 }
333         }
334
335         if (cloned_orig)
336                 kfree_skb(orig_skb);
337
338         STRP_STATS_ADD(strp->stats.bytes, eaten);
339
340         return eaten;
341 }
342
343 int strp_process(struct strparser *strp, struct sk_buff *orig_skb,
344                  unsigned int orig_offset, size_t orig_len,
345                  size_t max_msg_size, long timeo)
346 {
347         read_descriptor_t desc; /* Dummy arg to strp_recv */
348
349         desc.arg.data = strp;
350
351         return __strp_recv(&desc, orig_skb, orig_offset, orig_len,
352                            max_msg_size, timeo);
353 }
354 EXPORT_SYMBOL_GPL(strp_process);
355
356 static int strp_recv(read_descriptor_t *desc, struct sk_buff *orig_skb,
357                      unsigned int orig_offset, size_t orig_len)
358 {
359         struct strparser *strp = (struct strparser *)desc->arg.data;
360
361         return __strp_recv(desc, orig_skb, orig_offset, orig_len,
362                            strp->sk->sk_rcvbuf, strp->sk->sk_rcvtimeo);
363 }
364
365 static int default_read_sock_done(struct strparser *strp, int err)
366 {
367         return err;
368 }
369
370 /* Called with lock held on lower socket */
371 static int strp_read_sock(struct strparser *strp)
372 {
373         struct socket *sock = strp->sk->sk_socket;
374         read_descriptor_t desc;
375
376         if (unlikely(!sock || !sock->ops || !sock->ops->read_sock))
377                 return -EBUSY;
378
379         desc.arg.data = strp;
380         desc.error = 0;
381         desc.count = 1; /* give more than one skb per call */
382
383         /* sk should be locked here, so okay to do read_sock */
384         sock->ops->read_sock(strp->sk, &desc, strp_recv);
385
386         desc.error = strp->cb.read_sock_done(strp, desc.error);
387
388         return desc.error;
389 }
390
391 /* Lower sock lock held */
392 void strp_data_ready(struct strparser *strp)
393 {
394         if (unlikely(strp->stopped))
395                 return;
396
397         /* This check is needed to synchronize with do_strp_work.
398          * do_strp_work acquires a process lock (lock_sock) whereas
399          * the lock held here is bh_lock_sock. The two locks can be
400          * held by different threads at the same time, but bh_lock_sock
401          * allows a thread in BH context to safely check if the process
402          * lock is held. In this case, if the lock is held, queue work.
403          */
404         if (sock_owned_by_user_nocheck(strp->sk)) {
405                 queue_work(strp_wq, &strp->work);
406                 return;
407         }
408
409         if (strp->paused)
410                 return;
411
412         if (strp->need_bytes) {
413                 if (strp_peek_len(strp) >= strp->need_bytes)
414                         strp->need_bytes = 0;
415                 else
416                         return;
417         }
418
419         if (strp_read_sock(strp) == -ENOMEM)
420                 queue_work(strp_wq, &strp->work);
421 }
422 EXPORT_SYMBOL_GPL(strp_data_ready);
423
424 static void do_strp_work(struct strparser *strp)
425 {
426         read_descriptor_t rd_desc;
427
428         /* We need the read lock to synchronize with strp_data_ready. We
429          * need the socket lock for calling strp_read_sock.
430          */
431         strp->cb.lock(strp);
432
433         if (unlikely(strp->stopped))
434                 goto out;
435
436         if (strp->paused)
437                 goto out;
438
439         rd_desc.arg.data = strp;
440
441         if (strp_read_sock(strp) == -ENOMEM)
442                 queue_work(strp_wq, &strp->work);
443
444 out:
445         strp->cb.unlock(strp);
446 }
447
448 static void strp_work(struct work_struct *w)
449 {
450         do_strp_work(container_of(w, struct strparser, work));
451 }
452
453 static void strp_msg_timeout(struct work_struct *w)
454 {
455         struct strparser *strp = container_of(w, struct strparser,
456                                               msg_timer_work.work);
457
458         /* Message assembly timed out */
459         STRP_STATS_INCR(strp->stats.msg_timeouts);
460         strp->cb.lock(strp);
461         strp->cb.abort_parser(strp, ETIMEDOUT);
462         strp->cb.unlock(strp);
463 }
464
465 static void strp_sock_lock(struct strparser *strp)
466 {
467         lock_sock(strp->sk);
468 }
469
470 static void strp_sock_unlock(struct strparser *strp)
471 {
472         release_sock(strp->sk);
473 }
474
475 int strp_init(struct strparser *strp, struct sock *sk,
476               const struct strp_callbacks *cb)
477 {
478
479         if (!cb || !cb->rcv_msg || !cb->parse_msg)
480                 return -EINVAL;
481
482         /* The sk (sock) arg determines the mode of the stream parser.
483          *
484          * If the sock is set then the strparser is in receive callback mode.
485          * The upper layer calls strp_data_ready to kick receive processing
486          * and strparser calls the read_sock function on the socket to
487          * get packets.
488          *
489          * If the sock is not set then the strparser is in general mode.
490          * The upper layer calls strp_process for each skb to be parsed.
491          */
492
493         if (!sk) {
494                 if (!cb->lock || !cb->unlock)
495                         return -EINVAL;
496         }
497
498         memset(strp, 0, sizeof(*strp));
499
500         strp->sk = sk;
501
502         strp->cb.lock = cb->lock ? : strp_sock_lock;
503         strp->cb.unlock = cb->unlock ? : strp_sock_unlock;
504         strp->cb.rcv_msg = cb->rcv_msg;
505         strp->cb.parse_msg = cb->parse_msg;
506         strp->cb.read_sock_done = cb->read_sock_done ? : default_read_sock_done;
507         strp->cb.abort_parser = cb->abort_parser ? : strp_abort_strp;
508
509         INIT_DELAYED_WORK(&strp->msg_timer_work, strp_msg_timeout);
510         INIT_WORK(&strp->work, strp_work);
511
512         return 0;
513 }
514 EXPORT_SYMBOL_GPL(strp_init);
515
516 void strp_unpause(struct strparser *strp)
517 {
518         strp->paused = 0;
519
520         /* Sync setting paused with RX work */
521         smp_mb();
522
523         queue_work(strp_wq, &strp->work);
524 }
525 EXPORT_SYMBOL_GPL(strp_unpause);
526
527 /* strp must already be stopped so that strp_recv will no longer be called.
528  * Note that strp_done is not called with the lower socket held.
529  */
530 void strp_done(struct strparser *strp)
531 {
532         WARN_ON(!strp->stopped);
533
534         cancel_delayed_work_sync(&strp->msg_timer_work);
535         cancel_work_sync(&strp->work);
536
537         if (strp->skb_head) {
538                 kfree_skb(strp->skb_head);
539                 strp->skb_head = NULL;
540         }
541 }
542 EXPORT_SYMBOL_GPL(strp_done);
543
544 void strp_stop(struct strparser *strp)
545 {
546         strp->stopped = 1;
547 }
548 EXPORT_SYMBOL_GPL(strp_stop);
549
550 void strp_check_rcv(struct strparser *strp)
551 {
552         queue_work(strp_wq, &strp->work);
553 }
554 EXPORT_SYMBOL_GPL(strp_check_rcv);
555
556 static int __init strp_mod_init(void)
557 {
558         strp_wq = create_singlethread_workqueue("kstrp");
559
560         return 0;
561 }
562
563 static void __exit strp_mod_exit(void)
564 {
565         destroy_workqueue(strp_wq);
566 }
567 module_init(strp_mod_init);
568 module_exit(strp_mod_exit);
569 MODULE_LICENSE("GPL");