service_linux.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773
  1. package libnetwork
  2. import (
  3. "fmt"
  4. "io"
  5. "io/ioutil"
  6. "net"
  7. "os"
  8. "os/exec"
  9. "path/filepath"
  10. "runtime"
  11. "strconv"
  12. "strings"
  13. "sync"
  14. "syscall"
  15. "github.com/Sirupsen/logrus"
  16. "github.com/docker/docker/pkg/reexec"
  17. "github.com/docker/libnetwork/iptables"
  18. "github.com/docker/libnetwork/ipvs"
  19. "github.com/docker/libnetwork/ns"
  20. "github.com/gogo/protobuf/proto"
  21. "github.com/vishvananda/netlink/nl"
  22. "github.com/vishvananda/netns"
  23. )
  24. func init() {
  25. reexec.Register("fwmarker", fwMarker)
  26. reexec.Register("redirecter", redirecter)
  27. }
  28. // Get all loadbalancers on this network that is currently discovered
  29. // on this node.
  30. func (n *network) connectedLoadbalancers() []*loadBalancer {
  31. c := n.getController()
  32. c.Lock()
  33. serviceBindings := make([]*service, 0, len(c.serviceBindings))
  34. for _, s := range c.serviceBindings {
  35. serviceBindings = append(serviceBindings, s)
  36. }
  37. c.Unlock()
  38. var lbs []*loadBalancer
  39. for _, s := range serviceBindings {
  40. s.Lock()
  41. if lb, ok := s.loadBalancers[n.ID()]; ok {
  42. lbs = append(lbs, lb)
  43. }
  44. s.Unlock()
  45. }
  46. return lbs
  47. }
  48. // Populate all loadbalancers on the network that the passed endpoint
  49. // belongs to, into this sandbox.
  50. func (sb *sandbox) populateLoadbalancers(ep *endpoint) {
  51. var gwIP net.IP
  52. // This is an interface less endpoint. Nothing to do.
  53. if ep.Iface() == nil {
  54. return
  55. }
  56. n := ep.getNetwork()
  57. eIP := ep.Iface().Address()
  58. if n.ingress {
  59. if err := addRedirectRules(sb.Key(), eIP, ep.ingressPorts); err != nil {
  60. logrus.Errorf("Failed to add redirect rules for ep %s (%s): %v", ep.Name(), ep.ID()[0:7], err)
  61. }
  62. }
  63. if sb.ingress {
  64. // For the ingress sandbox if this is not gateway
  65. // endpoint do nothing.
  66. if ep != sb.getGatewayEndpoint() {
  67. return
  68. }
  69. // This is the gateway endpoint. Now get the ingress
  70. // network and plumb the loadbalancers.
  71. gwIP = ep.Iface().Address().IP
  72. for _, ep := range sb.getConnectedEndpoints() {
  73. if !ep.endpointInGWNetwork() {
  74. n = ep.getNetwork()
  75. eIP = ep.Iface().Address()
  76. }
  77. }
  78. }
  79. for _, lb := range n.connectedLoadbalancers() {
  80. // Skip if vip is not valid.
  81. if len(lb.vip) == 0 {
  82. continue
  83. }
  84. lb.service.Lock()
  85. for _, ip := range lb.backEnds {
  86. sb.addLBBackend(ip, lb.vip, lb.fwMark, lb.service.ingressPorts, eIP, gwIP, n.ingress)
  87. }
  88. lb.service.Unlock()
  89. }
  90. }
  91. // Add loadbalancer backend to all sandboxes which has a connection to
  92. // this network. If needed add the service as well.
  93. func (n *network) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig) {
  94. n.WalkEndpoints(func(e Endpoint) bool {
  95. ep := e.(*endpoint)
  96. if sb, ok := ep.getSandbox(); ok {
  97. if !sb.isEndpointPopulated(ep) {
  98. return false
  99. }
  100. var gwIP net.IP
  101. if ep := sb.getGatewayEndpoint(); ep != nil {
  102. gwIP = ep.Iface().Address().IP
  103. }
  104. sb.addLBBackend(ip, vip, fwMark, ingressPorts, ep.Iface().Address(), gwIP, n.ingress)
  105. }
  106. return false
  107. })
  108. }
  109. // Remove loadbalancer backend from all sandboxes which has a
  110. // connection to this network. If needed remove the service entry as
  111. // well, as specified by the rmService bool.
  112. func (n *network) rmLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, rmService bool) {
  113. n.WalkEndpoints(func(e Endpoint) bool {
  114. ep := e.(*endpoint)
  115. if sb, ok := ep.getSandbox(); ok {
  116. if !sb.isEndpointPopulated(ep) {
  117. return false
  118. }
  119. var gwIP net.IP
  120. if ep := sb.getGatewayEndpoint(); ep != nil {
  121. gwIP = ep.Iface().Address().IP
  122. }
  123. sb.rmLBBackend(ip, vip, fwMark, ingressPorts, ep.Iface().Address(), gwIP, rmService, n.ingress)
  124. }
  125. return false
  126. })
  127. }
  128. // Add loadbalancer backend into one connected sandbox.
  129. func (sb *sandbox) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, eIP *net.IPNet, gwIP net.IP, isIngressNetwork bool) {
  130. if sb.osSbox == nil {
  131. return
  132. }
  133. if isIngressNetwork && !sb.ingress {
  134. return
  135. }
  136. i, err := ipvs.New(sb.Key())
  137. if err != nil {
  138. logrus.Errorf("Failed to create an ipvs handle for sbox %s (%s,%s) for lb addition: %v", sb.ID()[0:7], sb.ContainerID()[0:7], sb.Key(), err)
  139. return
  140. }
  141. defer i.Close()
  142. s := &ipvs.Service{
  143. AddressFamily: nl.FAMILY_V4,
  144. FWMark: fwMark,
  145. SchedName: ipvs.RoundRobin,
  146. }
  147. if !i.IsServicePresent(s) {
  148. var filteredPorts []*PortConfig
  149. if sb.ingress {
  150. filteredPorts = filterPortConfigs(ingressPorts, false)
  151. if err := programIngress(gwIP, filteredPorts, false); err != nil {
  152. logrus.Errorf("Failed to add ingress: %v", err)
  153. return
  154. }
  155. }
  156. logrus.Debugf("Creating service for vip %s fwMark %d ingressPorts %#v in sbox %s (%s)", vip, fwMark, ingressPorts, sb.ID()[0:7], sb.ContainerID()[0:7])
  157. if err := invokeFWMarker(sb.Key(), vip, fwMark, ingressPorts, eIP, false); err != nil {
  158. logrus.Errorf("Failed to add firewall mark rule in sbox %s (%s): %v", sb.ID()[0:7], sb.ContainerID()[0:7], err)
  159. return
  160. }
  161. if err := i.NewService(s); err != nil && err != syscall.EEXIST {
  162. logrus.Errorf("Failed to create a new service for vip %s fwmark %d in sbox %s (%s): %v", vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
  163. return
  164. }
  165. }
  166. d := &ipvs.Destination{
  167. AddressFamily: nl.FAMILY_V4,
  168. Address: ip,
  169. Weight: 1,
  170. }
  171. // Remove the sched name before using the service to add
  172. // destination.
  173. s.SchedName = ""
  174. if err := i.NewDestination(s, d); err != nil && err != syscall.EEXIST {
  175. logrus.Errorf("Failed to create real server %s for vip %s fwmark %d in sbox %s (%s): %v", ip, vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
  176. }
  177. }
  178. // Remove loadbalancer backend from one connected sandbox.
  179. func (sb *sandbox) rmLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, eIP *net.IPNet, gwIP net.IP, rmService bool, isIngressNetwork bool) {
  180. if sb.osSbox == nil {
  181. return
  182. }
  183. if isIngressNetwork && !sb.ingress {
  184. return
  185. }
  186. i, err := ipvs.New(sb.Key())
  187. if err != nil {
  188. logrus.Errorf("Failed to create an ipvs handle for sbox %s (%s,%s) for lb removal: %v", sb.ID()[0:7], sb.ContainerID()[0:7], sb.Key(), err)
  189. return
  190. }
  191. defer i.Close()
  192. s := &ipvs.Service{
  193. AddressFamily: nl.FAMILY_V4,
  194. FWMark: fwMark,
  195. }
  196. d := &ipvs.Destination{
  197. AddressFamily: nl.FAMILY_V4,
  198. Address: ip,
  199. Weight: 1,
  200. }
  201. if err := i.DelDestination(s, d); err != nil && err != syscall.ENOENT {
  202. logrus.Errorf("Failed to delete real server %s for vip %s fwmark %d in sbox %s (%s): %v", ip, vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
  203. }
  204. if rmService {
  205. s.SchedName = ipvs.RoundRobin
  206. if err := i.DelService(s); err != nil && err != syscall.ENOENT {
  207. logrus.Errorf("Failed to delete service for vip %s fwmark %d in sbox %s (%s): %v", vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
  208. }
  209. var filteredPorts []*PortConfig
  210. if sb.ingress {
  211. filteredPorts = filterPortConfigs(ingressPorts, true)
  212. if err := programIngress(gwIP, filteredPorts, true); err != nil {
  213. logrus.Errorf("Failed to delete ingress: %v", err)
  214. }
  215. }
  216. if err := invokeFWMarker(sb.Key(), vip, fwMark, ingressPorts, eIP, true); err != nil {
  217. logrus.Errorf("Failed to delete firewall mark rule in sbox %s (%s): %v", sb.ID()[0:7], sb.ContainerID()[0:7], err)
  218. }
  219. }
  220. }
  221. const ingressChain = "DOCKER-INGRESS"
  222. var (
  223. ingressOnce sync.Once
  224. ingressProxyMu sync.Mutex
  225. ingressProxyTbl = make(map[string]io.Closer)
  226. portConfigMu sync.Mutex
  227. portConfigTbl = make(map[PortConfig]int)
  228. )
  229. func filterPortConfigs(ingressPorts []*PortConfig, isDelete bool) []*PortConfig {
  230. portConfigMu.Lock()
  231. iPorts := make([]*PortConfig, 0, len(ingressPorts))
  232. for _, pc := range ingressPorts {
  233. if isDelete {
  234. if cnt, ok := portConfigTbl[*pc]; ok {
  235. // This is the last reference to this
  236. // port config. Delete the port config
  237. // and add it to filtered list to be
  238. // plumbed.
  239. if cnt == 1 {
  240. delete(portConfigTbl, *pc)
  241. iPorts = append(iPorts, pc)
  242. continue
  243. }
  244. portConfigTbl[*pc] = cnt - 1
  245. }
  246. continue
  247. }
  248. if cnt, ok := portConfigTbl[*pc]; ok {
  249. portConfigTbl[*pc] = cnt + 1
  250. continue
  251. }
  252. // We are adding it for the first time. Add it to the
  253. // filter list to be plumbed.
  254. portConfigTbl[*pc] = 1
  255. iPorts = append(iPorts, pc)
  256. }
  257. portConfigMu.Unlock()
  258. return iPorts
  259. }
  260. func programIngress(gwIP net.IP, ingressPorts []*PortConfig, isDelete bool) error {
  261. addDelOpt := "-I"
  262. if isDelete {
  263. addDelOpt = "-D"
  264. }
  265. chainExists := iptables.ExistChain(ingressChain, iptables.Nat)
  266. filterChainExists := iptables.ExistChain(ingressChain, iptables.Filter)
  267. ingressOnce.Do(func() {
  268. // Flush nat table and filter table ingress chain rules during init if it
  269. // exists. It might contain stale rules from previous life.
  270. if chainExists {
  271. if err := iptables.RawCombinedOutput("-t", "nat", "-F", ingressChain); err != nil {
  272. logrus.Errorf("Could not flush nat table ingress chain rules during init: %v", err)
  273. }
  274. }
  275. if filterChainExists {
  276. if err := iptables.RawCombinedOutput("-F", ingressChain); err != nil {
  277. logrus.Errorf("Could not flush filter table ingress chain rules during init: %v", err)
  278. }
  279. }
  280. })
  281. if !isDelete {
  282. if !chainExists {
  283. if err := iptables.RawCombinedOutput("-t", "nat", "-N", ingressChain); err != nil {
  284. return fmt.Errorf("failed to create ingress chain: %v", err)
  285. }
  286. }
  287. if !filterChainExists {
  288. if err := iptables.RawCombinedOutput("-N", ingressChain); err != nil {
  289. return fmt.Errorf("failed to create filter table ingress chain: %v", err)
  290. }
  291. }
  292. if !iptables.Exists(iptables.Nat, ingressChain, "-j", "RETURN") {
  293. if err := iptables.RawCombinedOutput("-t", "nat", "-A", ingressChain, "-j", "RETURN"); err != nil {
  294. return fmt.Errorf("failed to add return rule in nat table ingress chain: %v", err)
  295. }
  296. }
  297. if !iptables.Exists(iptables.Filter, ingressChain, "-j", "RETURN") {
  298. if err := iptables.RawCombinedOutput("-A", ingressChain, "-j", "RETURN"); err != nil {
  299. return fmt.Errorf("failed to add return rule to filter table ingress chain: %v", err)
  300. }
  301. }
  302. for _, chain := range []string{"OUTPUT", "PREROUTING"} {
  303. if !iptables.Exists(iptables.Nat, chain, "-m", "addrtype", "--dst-type", "LOCAL", "-j", ingressChain) {
  304. if err := iptables.RawCombinedOutput("-t", "nat", "-I", chain, "-m", "addrtype", "--dst-type", "LOCAL", "-j", ingressChain); err != nil {
  305. return fmt.Errorf("failed to add jump rule in %s to ingress chain: %v", chain, err)
  306. }
  307. }
  308. }
  309. if !iptables.Exists(iptables.Filter, "FORWARD", "-j", ingressChain) {
  310. if err := iptables.RawCombinedOutput("-I", "FORWARD", "-j", ingressChain); err != nil {
  311. return fmt.Errorf("failed to add jump rule to %s in filter table forward chain: %v", ingressChain, err)
  312. }
  313. }
  314. oifName, err := findOIFName(gwIP)
  315. if err != nil {
  316. return fmt.Errorf("failed to find gateway bridge interface name for %s: %v", gwIP, err)
  317. }
  318. path := filepath.Join("/proc/sys/net/ipv4/conf", oifName, "route_localnet")
  319. if err := ioutil.WriteFile(path, []byte{'1', '\n'}, 0644); err != nil {
  320. return fmt.Errorf("could not write to %s: %v", path, err)
  321. }
  322. ruleArgs := strings.Fields(fmt.Sprintf("-m addrtype --src-type LOCAL -o %s -j MASQUERADE", oifName))
  323. if !iptables.Exists(iptables.Nat, "POSTROUTING", ruleArgs...) {
  324. if err := iptables.RawCombinedOutput(append([]string{"-t", "nat", "-I", "POSTROUTING"}, ruleArgs...)...); err != nil {
  325. return fmt.Errorf("failed to add ingress localhost POSTROUTING rule for %s: %v", oifName, err)
  326. }
  327. }
  328. }
  329. for _, iPort := range ingressPorts {
  330. if iptables.ExistChain(ingressChain, iptables.Nat) {
  331. rule := strings.Fields(fmt.Sprintf("-t nat %s %s -p %s --dport %d -j DNAT --to-destination %s:%d",
  332. addDelOpt, ingressChain, strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.PublishedPort, gwIP, iPort.PublishedPort))
  333. if err := iptables.RawCombinedOutput(rule...); err != nil {
  334. errStr := fmt.Sprintf("setting up rule failed, %v: %v", rule, err)
  335. if !isDelete {
  336. return fmt.Errorf("%s", errStr)
  337. }
  338. logrus.Infof("%s", errStr)
  339. }
  340. }
  341. // Filter table rules to allow a published service to be accessible in the local node from..
  342. // 1) service tasks attached to other networks
  343. // 2) unmanaged containers on bridge networks
  344. rule := strings.Fields(fmt.Sprintf("%s %s -m state -p %s --sport %d --state ESTABLISHED,RELATED -j ACCEPT",
  345. addDelOpt, ingressChain, strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.PublishedPort))
  346. if err := iptables.RawCombinedOutput(rule...); err != nil {
  347. errStr := fmt.Sprintf("setting up rule failed, %v: %v", rule, err)
  348. if !isDelete {
  349. return fmt.Errorf("%s", errStr)
  350. }
  351. logrus.Warnf("%s", errStr)
  352. }
  353. rule = strings.Fields(fmt.Sprintf("%s %s -p %s --dport %d -j ACCEPT",
  354. addDelOpt, ingressChain, strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.PublishedPort))
  355. if err := iptables.RawCombinedOutput(rule...); err != nil {
  356. errStr := fmt.Sprintf("setting up rule failed, %v: %v", rule, err)
  357. if !isDelete {
  358. return fmt.Errorf("%s", errStr)
  359. }
  360. logrus.Warnf("%s", errStr)
  361. }
  362. if err := plumbProxy(iPort, isDelete); err != nil {
  363. logrus.Warnf("failed to create proxy for port %d: %v", iPort.PublishedPort, err)
  364. }
  365. }
  366. return nil
  367. }
  368. // In the filter table FORWARD chain first rule should be to jump to INGRESS-CHAIN
  369. // This chain has the rules to allow access to the published ports for swarm tasks
  370. // from local bridge networks and docker_gwbridge (ie:taks on other swarm netwroks)
  371. func arrangeIngressFilterRule() {
  372. if iptables.ExistChain(ingressChain, iptables.Filter) {
  373. if iptables.Exists(iptables.Filter, "FORWARD", "-j", ingressChain) {
  374. if err := iptables.RawCombinedOutput("-D", "FORWARD", "-j", ingressChain); err != nil {
  375. logrus.Warnf("failed to delete jump rule to ingressChain in filter table: %v", err)
  376. }
  377. }
  378. if err := iptables.RawCombinedOutput("-I", "FORWARD", "-j", ingressChain); err != nil {
  379. logrus.Warnf("failed to add jump rule to ingressChain in filter table: %v", err)
  380. }
  381. }
  382. }
  383. func findOIFName(ip net.IP) (string, error) {
  384. nlh := ns.NlHandle()
  385. routes, err := nlh.RouteGet(ip)
  386. if err != nil {
  387. return "", err
  388. }
  389. if len(routes) == 0 {
  390. return "", fmt.Errorf("no route to %s", ip)
  391. }
  392. // Pick the first route(typically there is only one route). We
  393. // don't support multipath.
  394. link, err := nlh.LinkByIndex(routes[0].LinkIndex)
  395. if err != nil {
  396. return "", err
  397. }
  398. return link.Attrs().Name, nil
  399. }
  400. func plumbProxy(iPort *PortConfig, isDelete bool) error {
  401. var (
  402. err error
  403. l io.Closer
  404. )
  405. portSpec := fmt.Sprintf("%d/%s", iPort.PublishedPort, strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]))
  406. if isDelete {
  407. ingressProxyMu.Lock()
  408. if listener, ok := ingressProxyTbl[portSpec]; ok {
  409. if listener != nil {
  410. listener.Close()
  411. }
  412. }
  413. ingressProxyMu.Unlock()
  414. return nil
  415. }
  416. switch iPort.Protocol {
  417. case ProtocolTCP:
  418. l, err = net.ListenTCP("tcp", &net.TCPAddr{Port: int(iPort.PublishedPort)})
  419. case ProtocolUDP:
  420. l, err = net.ListenUDP("udp", &net.UDPAddr{Port: int(iPort.PublishedPort)})
  421. }
  422. if err != nil {
  423. return err
  424. }
  425. ingressProxyMu.Lock()
  426. ingressProxyTbl[portSpec] = l
  427. ingressProxyMu.Unlock()
  428. return nil
  429. }
  430. func writePortsToFile(ports []*PortConfig) (string, error) {
  431. f, err := ioutil.TempFile("", "port_configs")
  432. if err != nil {
  433. return "", err
  434. }
  435. defer f.Close()
  436. buf, err := proto.Marshal(&EndpointRecord{
  437. IngressPorts: ports,
  438. })
  439. n, err := f.Write(buf)
  440. if err != nil {
  441. return "", err
  442. }
  443. if n < len(buf) {
  444. return "", io.ErrShortWrite
  445. }
  446. return f.Name(), nil
  447. }
  448. func readPortsFromFile(fileName string) ([]*PortConfig, error) {
  449. buf, err := ioutil.ReadFile(fileName)
  450. if err != nil {
  451. return nil, err
  452. }
  453. var epRec EndpointRecord
  454. err = proto.Unmarshal(buf, &epRec)
  455. if err != nil {
  456. return nil, err
  457. }
  458. return epRec.IngressPorts, nil
  459. }
  460. // Invoke fwmarker reexec routine to mark vip destined packets with
  461. // the passed firewall mark.
  462. func invokeFWMarker(path string, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, eIP *net.IPNet, isDelete bool) error {
  463. var ingressPortsFile string
  464. if len(ingressPorts) != 0 {
  465. var err error
  466. ingressPortsFile, err = writePortsToFile(ingressPorts)
  467. if err != nil {
  468. return err
  469. }
  470. defer os.Remove(ingressPortsFile)
  471. }
  472. addDelOpt := "-A"
  473. if isDelete {
  474. addDelOpt = "-D"
  475. }
  476. cmd := &exec.Cmd{
  477. Path: reexec.Self(),
  478. Args: append([]string{"fwmarker"}, path, vip.String(), fmt.Sprintf("%d", fwMark), addDelOpt, ingressPortsFile, eIP.String()),
  479. Stdout: os.Stdout,
  480. Stderr: os.Stderr,
  481. }
  482. if err := cmd.Run(); err != nil {
  483. return fmt.Errorf("reexec failed: %v", err)
  484. }
  485. return nil
  486. }
  487. // Firewall marker reexec function.
  488. func fwMarker() {
  489. runtime.LockOSThread()
  490. defer runtime.UnlockOSThread()
  491. if len(os.Args) < 7 {
  492. logrus.Error("invalid number of arguments..")
  493. os.Exit(1)
  494. }
  495. var ingressPorts []*PortConfig
  496. if os.Args[5] != "" {
  497. var err error
  498. ingressPorts, err = readPortsFromFile(os.Args[5])
  499. if err != nil {
  500. logrus.Errorf("Failed reading ingress ports file: %v", err)
  501. os.Exit(6)
  502. }
  503. }
  504. vip := os.Args[2]
  505. fwMark, err := strconv.ParseUint(os.Args[3], 10, 32)
  506. if err != nil {
  507. logrus.Errorf("bad fwmark value(%s) passed: %v", os.Args[3], err)
  508. os.Exit(2)
  509. }
  510. addDelOpt := os.Args[4]
  511. rules := [][]string{}
  512. for _, iPort := range ingressPorts {
  513. rule := strings.Fields(fmt.Sprintf("-t mangle %s PREROUTING -p %s --dport %d -j MARK --set-mark %d",
  514. addDelOpt, strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.PublishedPort, fwMark))
  515. rules = append(rules, rule)
  516. }
  517. ns, err := netns.GetFromPath(os.Args[1])
  518. if err != nil {
  519. logrus.Errorf("failed get network namespace %q: %v", os.Args[1], err)
  520. os.Exit(3)
  521. }
  522. defer ns.Close()
  523. if err := netns.Set(ns); err != nil {
  524. logrus.Errorf("setting into container net ns %v failed, %v", os.Args[1], err)
  525. os.Exit(4)
  526. }
  527. if addDelOpt == "-A" {
  528. eIP, subnet, err := net.ParseCIDR(os.Args[6])
  529. if err != nil {
  530. logrus.Errorf("Failed to parse endpoint IP %s: %v", os.Args[6], err)
  531. os.Exit(9)
  532. }
  533. ruleParams := strings.Fields(fmt.Sprintf("-m ipvs --ipvs -d %s -j SNAT --to-source %s", subnet, eIP))
  534. if !iptables.Exists("nat", "POSTROUTING", ruleParams...) {
  535. rule := append(strings.Fields("-t nat -A POSTROUTING"), ruleParams...)
  536. rules = append(rules, rule)
  537. err := ioutil.WriteFile("/proc/sys/net/ipv4/vs/conntrack", []byte{'1', '\n'}, 0644)
  538. if err != nil {
  539. logrus.Errorf("Failed to write to /proc/sys/net/ipv4/vs/conntrack: %v", err)
  540. os.Exit(8)
  541. }
  542. }
  543. }
  544. rule := strings.Fields(fmt.Sprintf("-t mangle %s OUTPUT -d %s/32 -j MARK --set-mark %d", addDelOpt, vip, fwMark))
  545. rules = append(rules, rule)
  546. rule = strings.Fields(fmt.Sprintf("-t nat %s OUTPUT -p icmp --icmp echo-request -d %s -j DNAT --to 127.0.0.1", addDelOpt, vip))
  547. rules = append(rules, rule)
  548. for _, rule := range rules {
  549. if err := iptables.RawCombinedOutputNative(rule...); err != nil {
  550. logrus.Errorf("setting up rule failed, %v: %v", rule, err)
  551. os.Exit(5)
  552. }
  553. }
  554. }
  555. func addRedirectRules(path string, eIP *net.IPNet, ingressPorts []*PortConfig) error {
  556. var ingressPortsFile string
  557. if len(ingressPorts) != 0 {
  558. var err error
  559. ingressPortsFile, err = writePortsToFile(ingressPorts)
  560. if err != nil {
  561. return err
  562. }
  563. defer os.Remove(ingressPortsFile)
  564. }
  565. cmd := &exec.Cmd{
  566. Path: reexec.Self(),
  567. Args: append([]string{"redirecter"}, path, eIP.String(), ingressPortsFile),
  568. Stdout: os.Stdout,
  569. Stderr: os.Stderr,
  570. }
  571. if err := cmd.Run(); err != nil {
  572. return fmt.Errorf("reexec failed: %v", err)
  573. }
  574. return nil
  575. }
  576. // Redirecter reexec function.
  577. func redirecter() {
  578. runtime.LockOSThread()
  579. defer runtime.UnlockOSThread()
  580. if len(os.Args) < 4 {
  581. logrus.Error("invalid number of arguments..")
  582. os.Exit(1)
  583. }
  584. var ingressPorts []*PortConfig
  585. if os.Args[3] != "" {
  586. var err error
  587. ingressPorts, err = readPortsFromFile(os.Args[3])
  588. if err != nil {
  589. logrus.Errorf("Failed reading ingress ports file: %v", err)
  590. os.Exit(2)
  591. }
  592. }
  593. eIP, _, err := net.ParseCIDR(os.Args[2])
  594. if err != nil {
  595. logrus.Errorf("Failed to parse endpoint IP %s: %v", os.Args[2], err)
  596. os.Exit(3)
  597. }
  598. rules := [][]string{}
  599. for _, iPort := range ingressPorts {
  600. rule := strings.Fields(fmt.Sprintf("-t nat -A PREROUTING -d %s -p %s --dport %d -j REDIRECT --to-port %d",
  601. eIP.String(), strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.PublishedPort, iPort.TargetPort))
  602. rules = append(rules, rule)
  603. // Allow only incoming connections to exposed ports
  604. iRule := strings.Fields(fmt.Sprintf("-I INPUT -d %s -p %s --dport %d -m conntrack --ctstate NEW,ESTABLISHED -j ACCEPT",
  605. eIP.String(), strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.TargetPort))
  606. rules = append(rules, iRule)
  607. // Allow only outgoing connections from exposed ports
  608. oRule := strings.Fields(fmt.Sprintf("-I OUTPUT -s %s -p %s --sport %d -m conntrack --ctstate ESTABLISHED -j ACCEPT",
  609. eIP.String(), strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)]), iPort.TargetPort))
  610. rules = append(rules, oRule)
  611. }
  612. ns, err := netns.GetFromPath(os.Args[1])
  613. if err != nil {
  614. logrus.Errorf("failed get network namespace %q: %v", os.Args[1], err)
  615. os.Exit(4)
  616. }
  617. defer ns.Close()
  618. if err := netns.Set(ns); err != nil {
  619. logrus.Errorf("setting into container net ns %v failed, %v", os.Args[1], err)
  620. os.Exit(5)
  621. }
  622. for _, rule := range rules {
  623. if err := iptables.RawCombinedOutputNative(rule...); err != nil {
  624. logrus.Errorf("setting up rule failed, %v: %v", rule, err)
  625. os.Exit(6)
  626. }
  627. }
  628. if len(ingressPorts) == 0 {
  629. return
  630. }
  631. // Ensure blocking rules for anything else in/to ingress network
  632. for _, rule := range [][]string{
  633. {"-d", eIP.String(), "-p", "udp", "-j", "DROP"},
  634. {"-d", eIP.String(), "-p", "tcp", "-j", "DROP"},
  635. } {
  636. if !iptables.ExistsNative(iptables.Filter, "INPUT", rule...) {
  637. if err := iptables.RawCombinedOutputNative(append([]string{"-A", "INPUT"}, rule...)...); err != nil {
  638. logrus.Errorf("setting up rule failed, %v: %v", rule, err)
  639. os.Exit(7)
  640. }
  641. }
  642. rule[0] = "-s"
  643. if !iptables.ExistsNative(iptables.Filter, "OUTPUT", rule...) {
  644. if err := iptables.RawCombinedOutputNative(append([]string{"-A", "OUTPUT"}, rule...)...); err != nil {
  645. logrus.Errorf("setting up rule failed, %v: %v", rule, err)
  646. os.Exit(8)
  647. }
  648. }
  649. }
  650. }