/usr/share/pyshared/pyroman/port.py is in pyroman 0.5.0-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | #Copyright (c) 2011 Erich Schubert erich@debian.org
#Permission is hereby granted, free of charge, to any person obtaining a copy
#of this software and associated documentation files (the "Software"), to deal
#in the Software without restriction, including without limitation the rights
#to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#copies of the Software, and to permit persons to whom the Software is
#furnished to do so, subject to the following conditions:
#The above copyright notice and this permission notice shall be included in
#all copies or substantial portions of the Software.
#THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
#SOFTWARE.
import re, socket
from exception import PyromanException
class PortInvalidSpec(Exception):
"""
Exception class for invalid port specifications
"""
def __init__(self, err, spec):
"""
Create new InvalidPortSpecification
"""
self.err = err
self.spec = spec
def __str__(self):
"""
Return error message
"""
return self.err
class Port:
"""
This class represents a single tcp/udp/icmp port
(or the whole protocol!)
It's instantiated from a string, which is parsed and checked
and it contains a to_filter method to generate an iptables filter
"""
# Split and verify syntax of statement
preg = re.compile("^(?:([a-z0-9\-]+|[0-9]+(?:\:[0-9]+)?)(?:/))?(tcp|udp|esp|ah|icmp|icmpv6|ipv6-icmp|esp|ah)$")
# verify port range
prreg = re.compile("^([0-9]+:)?[0-9]+$")
def __init__(self, spec):
"""
Initialize port from a specification string of the type "123/tcp"
If the string is not parseable, PortInavlidSpec is raised
"""
self.proto = ""
self.port = None
# if a spec is given, process
if spec != "":
m = self.preg.match( spec )
if m is None:
raise PortInvalidSpec("Invalid port specification: %s" % spec, spec)
self.port = m.group(1)
self.proto = m.group(2)
# if it's a named port, verify it's resolveable...
if self.proto in ["udp", "tcp"] and not self.prreg.match(self.port):
try:
socket.getservbyname(self.port, self.proto)
except socket.error:
raise PortInvalidSpec("Port %s/%s not defined in /etc/services" % (self.port, self.proto), spec)
def get_filter_proto(self):
"""
Return iptables rule to filter for this protocol as string
"""
if self.proto:
return "-p %s" % self.proto
def get_filter_port(self, dir):
"""
Return iptables rule to filter for this specific port as string
An appropriate protocol filter needs to be added, too (use the
get_filter_proto method for that). Note that source and destination
filters only make sense if they use the same protocol!
dir -- direction ("d" for destination or "s" for source filter)
"""
if not self.port:
return ""
if self.proto in ["tcp", "udp"]:
return "--" + dir + "port " + self.port
elif self.proto == "icmp":
# ICMP doesn't have source ports
if dir == "d":
return "--icmp-type " + self.port
else:
return ""
elif self.proto in [ "icmpv6", "ipv6-icmp"]:
# ICMP doesn't have source ports
if dir == "d":
return "--icmpv6-type " + self.port
else:
return ""
elif self.proto in ("esp", "ah"):
# no port for ESP and AH
return ""
else:
raise PyromanException("Unknown protocol: %s" % self.proto)
def forIPv4(self):
"""
Return true when compatible with IPv4
"""
return self.proto not in ["icmpv6", "ipv6-icmp"]
def forIPv6(self):
"""
Return true when compatible with IPv6
"""
return self.proto not in ["icmp", "ah"]
|