service_linux.go 22 KB

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