/usr/lib/python3/dist-packages/fail2ban/tests/__pycache__
Edit: /usr/lib/python3/dist-packages/fail2ban/tests/__pycache__/utils.cpython-310.pyc (25270B)
o
;s*bv @ s d Z dZdZddlZddlZddlZddlZddlZddlZddl Z ddl
Z
ddl Z ddl Z ddl
Z
ddlZddlmZ ddlmZ ddlmZmZmZmZ dd lmZmZ dd
lm Z dd l m Z dd l m Z dd
l!m!Z! ee"a#dZ$ej%&ddZ'e'sej()ej(*ddrdZ'ndZ'ej()ej(*e'dZ+e,dej(-ej(-ej(-ej(.e/ G dd dej0Z1dCddZ2dd Z3G dd d e1Z4d d Z5d d Z6e7ed!sG d"d! d!e8Z9e9e_9ej:j;Z
ej%&d)dZ?d*d+ Z@d,d- ZAdDd.d/ZBddlCZCddlDZDe7ejEd0s dEd1d0ZFeFejE_Fd2d3eGdfd4d5ZHeHejE_He7ejEd6s ZOejEjPZQd?d@ ZReOejE_MeRejE_PG dAdB dBejEZSe jTZTdS )FzYaroslav Halchenkoz%Copyright (c) 2013 Yaroslav HalchenkoGPL N)StringIOwraps ) getLogger str2LogLevelgetVerbosityFormat
uni_decode)IPAddrDNSUtils)MyTime)Utils) asyncserver)versioni BFAIL2BAN_CONFIG_DIRconfigz
fail2ban.confz
/etc/fail2ban
PYTHONPATHc @ s e Zd Zdd ZdS )DefaultTestOptionsc C s d d dd d dddddd
| _ d S )NTF)
log_level verbositylog_lazy
log_tracebackfull_tracebackfast memory_dbno_gamin
no_network negate_re)__dict__self r" 6/usr/lib/python3/dist-packages/fail2ban/tests/utils.py__init__H s
zDefaultTestOptions.__init__N)__name__
__module__ __qualname__r$ r" r" r" r# r G s r c C s t j}t jdtjd | dt d}||dddd dd |d
d d d d
d|ddd td dd|dddddd|ddddd d |d d dd d!d |d"d#dd$d%d |d&d'dd(d)d |d*d+dd,d-d |d.d/dd0d1|d2dd3d1g |S )4Nz%s [OPTIONS] [regexps]
r z%prog )usager z-lz --log-levelr z4Log level for the logger to use during running tests)destdefaulthelpz-vcountr zIncrease verbosity)actionr* r+ r, z --verbositystorez'Set numerical level of verbosity (0..4))r. r* typer+ r, z --log-direct store_falser Tz!Prevent lazy logging inside testsz-nz --no-network
store_truer z)Do not run tests that require the network)r. r* r, z-gz
--no-gaminr z'Do not run tests that require the gaminz-mz --memory-dbr z/Run database tests using memory instead of filez-fz--fastr zQTry to increase speed of the tests, decreasing of wait intervals, memory databasez-iz--ignorer zAnegate [regexps] filter to ignore tests matched specified regexpsz-tz--log-tracebackz.Enrich log-messages with compressed tracebacks)r. r, z--full-tracebackzBEither to make the tracebacks full, not compressed (as by default))optparseOption OptionParsersysargvr add_optionsint)docr4 pr" r" r# getOptParserS sh
#r< c C s6 t dad }| jd urt| j}t| nttj tj| _| j}|d u rV|d u r-dn%|tj kr4dn |tj
kr;dn|t tj tj
krFdn |t tjtjkrQdnd}|| _ttj}d}| jrqddlm} | jrld pmd
| }ntj}t||}||| t| | jdkrtd t t tj d d
f | S )Nfail2ban r r z %(message)s)FormatterWithTraceBackz %(tb)sz %(tbc)sz1Fail2ban %s test suite. Python %s. Please wait...
r( ) r logSysr r setLevelloggingCRITICALlevelr
HEAVYDEBUGDEBUGminINFONOTICEWARNINGERROR
StreamHandlerr6 stdoutr helpersrA r Formatterr setFormatter
addHandlerprintr strreplace)optsllevr rP fmtrR r" r" r# initProcess s>
r[ c @ sF e Zd ZdZdZdd Zdd Zdd Zd d
Zd d Z dddZ
dS )F2B< c C s( |j | _ | jr
d| _d| _i | j d< d S )NT share_config)r r r r )r! rX r" r" r# r$ s
z F2B.__init__c C d S Nr" r r" r" r#
SkipIfFast zF2B.SkipIfFastc C r` ra r" r r" r" r# SkipIfNoNetwork rc zF2B.SkipIfNoNetworkc K s t s@|dr tddD ]3}||}|du rqtj|d dkr(|d7 }tjtjt |d |s?td ||f qdS dS )
z4Helper to check action/filter config is available
stockz/Skip test because of missing stock-config files)r. filterNr> r( z.confz.dz-Skip test because of missing %s-config for %r)
STOCKgetunittestSkipTestospathsplitextexistsjoin
CONFIG_DIR)r! kwargstvr" r" r# SkipIfCfgMissing s
zF2B.SkipIfCfgMissingc s fdd}|S )z>Helper decorator to check action/filter config is available
c s t fdd}|S )Nc s( t jjdi | g|R i |S )Nr" )ri r\ rt r! argsrq )decargsfr" r# wrapper s z?F2B.skip_if_cfg_missing.._deco_wrapper..wrapperr rx ry rw rx r#
_deco_wrapper s z.F2B.skip_if_cfg_missing.._deco_wrapperr" )r! rw r} r" r{ r# skip_if_cfg_missing s zF2B.skip_if_cfg_missingTc C s4 t |tr
|r
| j}| jrt |trt|d }|S )N
)
isinstancebool MAX_WAITTIMEr r9 float)r! wtimer" r" r# maxWaitTime s
zF2B.maxWaitTimeNT) r% r& r' r MID_WAITTIMEr$ rb rd rt r~ r r" r" r" r# r\ s
r\ c t fdd}|S )zHelper decorator to create a temporary directory
Directory gets removed after function returns, regardless
if exception was thrown of not
c s> t jdd}z | |g|R i |W t| S t| w )Nzf2b-temp)prefix)tempfilemkdtempshutilrmtree)r! rv rq tmpr| r" r# ry s z with_tmpdir..wrapperr rz r" r| r# with_tmpdir s r c r )z@Helper decorator to execute test in alternate (fixed) test time.c s. t z | g|R i |W t S t w ra ) setUpMyTimetearDownMyTimeru r| r" r# ry s z with_alt_time..wrapperr rz r" r| r#
with_alt_time s r rj c @ s e Zd ZdS )rj N)r% r& r' r" r" r" r# rj s c C sZ |d t u r%| jr| jt|d d S | jr#| jd | j d S t| || d S )Nr r> s) rj showAllstreamwritelnrV dotswriteflush
_org_AddError)r! testerrr" r" r# addError s
r c s | st t } t| t_tjjr"dt_dt_dt_dd }|tj_ ndt_dt_dt_t
j fdd }|t
_ tjj rCd
d }|tj_
d tjd
< t
tt tj}|jddd tdD ]"}|d| d |d| d |d| d |d| d q_|dd |dd tjj r|dd |dd |dd tj}d t fd t fd t fd t fd!t fd"t fd#ttd$td%gfd&ttd$td%gffD ]}|j| qtjjrt D ]}||t d' qd S d S d S )(Ng{Gzd?gMb@?g-C 6?c S
t d)Nz Skip test because of "--fast"ri rj r" r" r" r# F2B_SkipIfFast"
z!initTests..F2B_SkipIfFastg?g{Gzt?c s | dkr
t d| | d S )Ng ?zT[BAD-CODE] To long sleep interval: %s, try to use conditional Utils.wait_for instead)
ValueErrorrs
_org_sleepr" r#
_new_sleep, s z initTests.._new_sleepc S r )Nz#Skip test because of "--no-network"r r" r" r" r# F2B_SkipIfNoNetwork3 r z&initTests..F2B_SkipIfNoNetworkz CET-01CEST-02,M3.5.0,M10.5.0TZi' i, )maxCountmaxTime z
192.0.2.%sz
198.51.100.%sz 203.0.113.%sz 2001:db8::%sz test-hostz2001:db8::ffffz
test-otherz
87.142.124.10z 192.0.2.888z8.8.4.4z
dns.googlez999.999.999.999z
abcdef.abcdefz
192.168.0.z
failed.dns.chz!doh1.2.3.4.buga.xxxxx.yyy.invalidz 1.2.3.4.buga.xxxxx.yyy.invalidz example.comz"2606:2800:220:1:248:1893:25c8:1946z
93.184.216.34zwww.example.com localhost) r[ r r\ ri r r DEFAULT_SLEEP_TIMEDEFAULT_SLEEP_INTERVALDEFAULT_SHORT_INTERVALrb timesleepr rd rk environtzsetr
setAlternateNowTEST_NOWr CACHE_ipToName
setOptionsrangesetCACHE_nameToIpr getSelfNamesdnsToIp)rX r r r cir" r r# initTests sf
r c C r` ra r" r" r" r" r#
mtimesleepe s r r c C s t t d S ra )r
setTimer r" r" r" r# r m s r c C s
d t _d S ra )r
myTimer" r" r" r# r s r r c
s" t ddlm} ddlm} ddlm} ddlm} ddlm} ddlm} ddlm} dd lm } dd
lm
}
dd lm } dd lm } dd
lm
}
ddlm} ddlm} ddlm} ddlm} ddlm} sqt }nG fdddtj }|t|j |t|j |t|j |t|j |t|j |t|
j |t| j |t|j |t|j |t|j |t|j zdd l!}|t|j" W n t#y Y nw |t|j$ |t|j% |t|j& |t|j' |t|j( |t|j) |t| j* |t| j+ |t|
j, |t|
j- |t|
j. |t|
j/ |t|j0 |t|j1 |t|j2 |t|j3 |t|j4 |t|j5 |t|j6 |t|j7 |t|j8 |t|j9 |t|j: |t|j; |t|j< |t|j |t| j= |t| j> |t|j? |t|j@ |t|jA |t|jB tC }ddlmD} tEFtEjGHtEjGI|jJD ]"}|KdrV|LdrV||Md|jNtEjGO|d f q5ddlPmQ} |g}z#tjRjSsltjRjTrxt#d tjRjStjRjTf dd lUmV} |W| W n t#y } z
tXYd | W Y d }~nd }~ww z
dd lZm[} |W| W n t#y } z
tXYd | W Y d }~nd }~ww |D ]} |t|\| qzdd!l]m^} |t|_| W n t#y } z
tXYd"| W Y d }~nd }~ww |t|j` |S )#Nr> )banmanagertestcase)clientbeautifiertestcase)clientreadertestcase)tickettestcase)failmanagertestcase)filtertestcase)servertestcase)datedetectortestcase)actiontestcase)actionstestcase)sockettestcase) misctestcase)databasetestcase)observertestcase)samplestestcase)fail2banclienttestcase)fail2banregextestcasec s. e Zd Zdd D Z fddZ ZS )z&gatherTests..FilteredTestSuitec S s g | ]}t |qS r" )recompile).0rr" r" r#
s z1gatherTests..FilteredTestSuite.c s g }|D ],}t |tjr| | qt|}| jD ]}||}js&|r/n|s/|| nqq|D ]
}t | | q3d S ra )
r ri TestSuiteaddTestrV _regexpssearchr appendsuper)r! suitematchedr r r m)FilteredTestSuite __class__rX r" r# r s
z.gatherTests..FilteredTestSuite.addTest)r% r& r' r r
__classcell__r" r rX regexpsr r# r s r r )action_dtest_z.pyz%s.%sr )
FilterPollz Skip, fast: %s, no_gamin: %s) FilterGaminz2Skipping gamin backend testing. Got exception '%s')FilterPyinotifyz9I: Skipping pyinotify backend testing. Got exception '%s')
FilterSystemdz7I: Skipping systemd backend testing. Got exception '%s')ar r( r r r r r r r r r r r r r r r r r ri r r makeSuite Transmitter JailTests
RegexTests LoggingTestsServerConfigReaderTestsCommandActionTestExecuteActions TicketTests
AddFailureFailmanagerComplexdnsStatusExtendedCymruInfo ImportErrorBeautifierTestConfigReaderTestJailReaderTestFilterReaderTestJailsReaderTestJailsReaderTestCacheSocket
ClientMisc HelpersTest SetupTestTestsUtilsTest
MyTimeTest DatabaseTest ObserverTest BanTimeIncr
BanTimeIncrDBIgnoreIP BasicFilterLogFileLogFileMonitorLogFileFilterPoll IgnoreIPDNS GetFailures
DNSUtilsTestsDNSUtilsNetworkTestsDateDetectorTestCustomDateFormatsTestFilterSamplesRegexFail2banClientTestFail2banServerTestFail2banRegexTest
TestLoaderr rk listdirrl abspathdirname__file__
startswithendswithloadTestsFromNamer% rm server.filterpollr r\ r r server.filtergaminr r rC warningserver.filterpyinotifyr get_monitor_failures_testcaseserver.filtersystemdr %get_monitor_failures_journal_testcaseTransmitterLogging) r rX r r r r r r r r r r r r r r r r r testsr
testloaderr file_r filtersr er Filter_r r" r r# gatherTestsw s
r1 assertDictEqualc C s | t|td | t|td ||kr>d||f }ddtt| t| }|p6|| }| | d S d S )Nz"First argument is not a dictionaryz#Second argument is not a dictionary%r != %rrB )
assertTruer dictro difflibndiffpprintpformat
splitlinesfail)r! d1d2msg standardMsgdiffr" r" r# r2 ' s
r> Fc
s dd |r t n|ffdd fdd z ||||| W d
S tyg } z7t|tr6|jd nt|d }d d tt
| t
| } |pV|| }| | W Y d
}~d
S d
}~ww ) zCompare complex elements (like dict, list or tuple) in sorted order until
level 0 not reached (initial level = -1 meant all levels),
or if nestedOnly set to True and some of the objects still contains nested lists or dicts.
c S s2 t | trtdd | D S tdd | D S )Nc s | ] }t |tttfV qd S ra r r5 listtupler rs r" r" r# ; z8assertSortedEqual.._is_nested..c s rA ra rB rE r" r" r# rF < rG )r r5 anyvaluesr r" r" r#
_is_nested9 s
z%assertSortedEqual.._is_nestedc s4 t | tttfrtt fdd| D dS | S )Nc 3 s | ]} |V qd S ra r" rE ) _nest_sortedkeyr" r# rF B s z:assertSortedEqual.._nest_sorted..rL )r r rC rD sorted)rs rL )rK rM r# rK @ s z'assertSortedEqual.._nest_sortedc sj t | t |krtd| |f |s*|r*| s*|s*| |kr"d S td| |f t| trpt|trp| D ]5\}}|| }t|tttfrat|tttfra |||dkr[|d nd|| q8||krmtd| |f q8d S | |d} ||d}t| |D ]1\}}t|tttfrt|tttfr |||dkr|d nd|| q||krtd| |f qd S )Nr3 r r> rM )lenr r r5 itemsrC rD zip)abrG
nestedOnlyrL kv1v2_assertSortedEqualrJ rK r" r# rY E s0 " "z-assertSortedEqual.._assertSortedEqualr z
within:rB N)
rN Exceptionr r rv rV ro r6 r7 r8 r9 r: r; )
r! rR rS rG rT rL r> r/ r? r@ r" rX r# assertSortedEqual3 s"
r[ assertRaisesRegexpc
O s z ||i | W n- |y6 } z!t |t|d u r+| d||f W Y d }~d S W Y d }~d S d }~ww | dt|d d S )Nz"%s" does not match "%s"z
%s not raisedr% )r r rV r; getattr)r! excclsregexpfunrv rq r/ r" r" r# r\ j s Tc C sl |}d}|d u rt |drt|tst|\}}d}||vr4|r%t|}|p,d||f }| | d S d S )NF__iter__Tz%r was not found in %rhasattrr rV itertoolsteerC r; r! rR rS r> bbwrapr" r" r# assertInv ri c C sl |}d}|d u rt |drt|tst|\}}d}||v r4|r%t|}|p,d||f }| | d S d S )NFra Tz%r unexpectedly found in %rrb rf r" r" r# assertNotIn rj rk c C sP t jjtjkrtjd t d|
t | t jj dkr&t
| _d S d S )NrB z"========== %s ====================r )ri r\ r rE rI r6 stderrr rC debugid
_org_setUpr r __startTimer r" r" r# _customSetUp s rq c C s0 t jjdkrtjdt | j f d S d S )Nr z
%.3fs -- )ri r\ r r6 rl r r rp r r" r" r# _customTearDown s rr c sv e Zd ZG dd dejZ fddZ fddZdd Zd d
Z d d Z
dddZ dd Z e
ejfddZ ZS )LogCaptureTestCasec @ sD e Zd ZdZdddZdddZdd Zd
d Zd d
Zdd Z dS )z LogCaptureTestCase._MemHandlerzLogging handler helper
Affords not to delegate logging to StreamHandler at all,
format lazily on demand in getvalue.
Increases performance inside the LogCaptureTestCase tests, because there
the log level set to DEBUG.
Tc C sL t | _d| _d| _t | _d| _t | _ t
j | |r$| j
| _d S d S )Nr( r ) threadingLock_lock_val_dirtyrC _recs
_nolckCntrr _strmrE Handlerr$ _handle_lazyhandle)r! lazyr" r" r# r$ s
z'LogCaptureTestCase._MemHandler.__init__Nc C s\ |rt d| d| _| j d| _t | _| jd W d dS 1 s'w Y dS )z)Truncate the internal buffer and records.z.invalid size argument: %r, should be None or 0r( r N)rZ rw rv rx rC ry r{ truncate)r! sizer" r" r# r s "z'LogCaptureTestCase._MemHandler.truncatec
C s z&| d }z
| j| W W d S ty& | j|dd Y W d S w tyA } z| jd| W Y d }~d S d }~ww )NrB zUTF-8rW z Error by logging handler: %r)
getMessager{ r UnicodeEncodeErrorencoderZ )r! recordr> r/ r" r" r# __write s z&LogCaptureTestCase._MemHandler.__writec C s | j s| jS | jd}| j d@ rQ|s*| jd7 _| jdkr"| jS d| _| j z| j dM _ | j}t | _W | j n| j w |D ]}| | qHn|r_| j dM _ | j | j
| _| jS )z&Return current buffer as whole string.Fr r> r ) rx rw rv acquirerz ry rC release_MemHandler__writer{ getvalue)r! lckrecsr r" r" r# r s.
z'LogCaptureTestCase._MemHandler.getvaluec C sD | | | j | jdO _W d dS 1 sw Y dS )z-Handle the specified record direct (not lazy)r> N)r rv rx r! r r" r" r# r~ s
"z%LogCaptureTestCase._MemHandler.handlec C sF | j | j| | jdO _W d dS 1 s w Y dS )z*Lazy handle the specified record on demandr N)rv ry r rx r r" r" r# r} s "z+LogCaptureTestCase._MemHandler._handle_lazyr ra )
r% r& r' __doc__r$ r r r r~ r} r" r" r" r# _MemHandler s
" r c sx t j| _t j| _ttjj | _
| j
gt _| jt j kr#t j| j7 _| jt j d kr3t
t j d tt| d S )Nr> )rC rG
_old_levelhandlers
_old_handlersrs r ri r\ r _logrE rI rD r setUpr r r" r# r s
zLogCaptureTestCase.setUpc s* | | jt_| jt_tt| dS )zCall after every test case.N) pruneLogr rC r r rG r rs tearDownr r r" r# r s zLogCaptureTestCase.tearDownc O sP | j }|dds|D ] }||v r dS q
dS |D ] }||vr% dS q dS )NallFT)r r rh r! r rq loggeds_r" r" r#
_is_logged s
z LogCaptureTestCase._is_loggedc s dd}|rtj|}t fdd|}nji } ddsD|sBj } d|r:d| nd |f dS dS |sdj }D ]}||vrc d
||r]d| nd |f qMdS dS ) ak Assert that one of the strings was logged
Preferable to assertTrue(self._is_logged(..)))
since provides message with the actual log.
Parameters
----------
s : string or list/set/tuple of strings
Test should succeed if string (or any of the listed) is present in the log
all : boolean (default False) if True should fail if any of s not logged
waitNc s j i S ra )r r" rq r r! r" r# 2 s z1LogCaptureTestCase.assertLogged..r Fz/None among %r was found in the log%s: ===
%s===z , waited %sr( z(%r was not found in the log%s: ===
%s===)
rh ri r\ r r wait_forr r r r; )r! r rq r resr r r" r r# assertLogged# s,
z LogCaptureTestCase.assertLoggedc O sx | j }t|dkr(|dds(|D ] }||vr dS q| d||f dS |D ]}||v r9| d||f q*dS )a Assert that strings were not logged
Parameters
----------
s : string or list/set/tuple of strings
Test should succeed if the string (or at least one of the listed) is not
present in the log
all : boolean (default False) if True should fail if any of s logged
r> r FNz6All of the %r were found present in the log: ===
%s===z"%r was found in the log: ===
%s===)r r rO rh r; r r" r" r# assertNotLoggedD s
z"LogCaptureTestCase.assertNotLoggedNc C s$ | j d |rtd| d S d S )Nr z===== %s =====)r r rC rm )r! logphaser" r" r# r Y s zLogCaptureTestCase.pruneLogc C s
| j S ra )r r r r" r" r# getLog^ r zLogCaptureTestCase.getLogc C sZ |t jks t tjkr+|d| d t| D ] }|d}|| q|d dS dS )z@Helper which outputs content of the file at HEAVYDEBUG loglevelsz---- z ----rB z ------------------------------N)rC rm getEffectiveLevelrE rI fileinputinputrstrip)fnr~ liner" r" r# dumpFilea s
zLogCaptureTestCase.dumpFilera )r% r& r' rE r| r r r r r r r r staticmethodrC rm r r r" r" r r# rs s X !
rs )r( )NNra )U
__author__
__copyright__ __license__r rd rE r3 rk r r r r6 r rt ri ior functoolsr rQ r r r r
server.ipdnsr r
server.mytimer
server.utilsr serverr r r% rC r r rh rp rl rn ro rg putenvr r r Valuesr r< r[ r\ r r rc rZ rj _TextTestResultr r r r old_TZr r r1 r6 r8 TestCaser2 reprr[ r\ assertRaisesRegexri rk r ro rq r
_org_tearDownrr rs
pid_existsr" r" r" r# s
,82
O
/
4
L