strparser.c 13 KB

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