snstac / pytak

PyTAK is a Python package for rapid TAK integration.

Python Team Awareness Kit (PyTAK)

PyTAK is a Python Module for creating TAK clients, servers & gateways.

PyTAK includes:

• TAK Protocol support for interconnecting with ATAK, WinTAK, iTAK & TAK Server.
• Classes for handling TAK, Cursor on Target (CoT) & non-CoT data.
• Functions for parsing and serializing TAK & CoT data.
• Functions for sending and receiving TAK & CoT data over a network.

Documentation

See PyTAK documentation for instructions on getting started with PyTAK, examples, configuration & troubleshooting options.

License & Copyright

Copyright Sensors & Signals LLC https://www.snstac.com

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

asyncio_dgram is Copyright (c) 2019 Justin Bronder and is licensed under the MIT License, see pytak/asyncio_dgram/LICENSE for details.

pytak's Issues

Add config params MAX_OUT_QUEUE & MAX_IN_QUEUE to allow queue tuning.

Add config params MAX_OUT_QUEUE & MAX_IN_QUEUE to allow queue tuning.

Add method to define trusted partner TLS certs

We can avoid ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate in certificate chain (_ssl.c:1131) with bypass settings such as PYTAK_TLS_DONT_CHECK_HOSTNAME.

What are the instructions for adding hosts to our trust chain?

PyTAK Multicast read/write & write-only do not work on Windows

re: https://stackoverflow.com/questions/15322242/python-bind-error-on-multicast-bind-on-windows

Write-only multicast doesn't work either for the same reason.

Help with tls certificate and connecting to TAK server

Hi I am having issues using the your library. I would like to send coordinate with CoT to a TAK Server. I have an up and runing server on a raspberry pi setup with the tls certificate but when I try to run the tak_pong example, I receive the following error :
Exception: Please specify COT_URL as a full URL, including '://', for example: tcp://tak.example.com:1234 .

Do you have an idea why I get this error ? The only thing I change is to add the require part to make it work with the tls certificates.

Thank in advance for your help.

#!/usr/bin/env python3

import asyncio
import xml.etree.ElementTree as ET

from configparser import ConfigParser

import pytak


class MySerializer(pytak.QueueWorker):
    """
    Defines how you process or generate your Cursor-On-Target Events.
    From there it adds the COT Events to a queue for TX to a COT_URL.
    """

    async def handle_data(self, data):
        """
        Handles pre-COT data and serializes to COT Events, then puts on queue.
        """
        event = data
        await self.put_queue(event)

    async def run(self, number_of_iterations=-1):
        """
        Runs the loop for processing or generating pre-COT data.
        """
        while 1:
            data = tak_pong()
            await self.handle_data(data)
            await asyncio.sleep(20)


def tak_pong():
    """
    Generates a simple takPong COT Event.
    """
    root = ET.Element("event")
    root.set("version", "2.0")
    root.set("type", "t-x-d-d")
    root.set("uid", "takPong")
    root.set("how", "m-g")
    root.set("time", pytak.cot_time())
    root.set("start", pytak.cot_time())
    root.set("stale", pytak.cot_time(3600))
    return ET.tostring(root)


async def main():
    """
    The main definition of your program, sets config params and
    adds your serializer to the asyncio task list.
    """
    config = ConfigParser()
    config["mycottool"] = {"COT_URL": "tcp://192.168.0.115:8089"}
    config["mycottool"] = {"PYTAK_TLS_CLIENT_CERT": "user.pem"}
    config = config["mycottool"]
    

    # Initializes worker queues and tasks.
    clitool = pytak.CLITool(config)
    await clitool.setup()

    # Add your serializer to the asyncio task list.
    clitool.add_tasks(set([MySerializer(clitool.tx_queue, config)]))

    # Start all tasks.
    await clitool.run()


if __name__ == "__main__":
    asyncio.run(main())

Allow sub-second backoff sleep

https://github.com/ampledata/pytak/blob/4e8ad2377728b9ca34cf3d164b142d2fef4a20e5/pytak/classes.py#L59

Currently this code-block casts any sleep period to an int, which restricts us to either 0 sleep, or sleep >= 1 second. For more variability, as well as sub-second sleep times, this should be refactored to allow float and not force to int.

Add additional default CoT attributes

ACCESS, CAVEAT, RELTO, QOS, OPEX, VAL, etc.

Add a way of disabling multicast receive

Yes, I realize I just added multicast receive, but running more than one multicast client on the same host is blocked.

The use case is: I want to run multiple multicast senders on one machine.

Add tests for Python 3.11 & 3.12

Default constant types are wrong

Default constants (e.g. DEFAULT_TAK_PROTO: int = 0) have the wrong Type Hints. Most of these constants are used as default parameters for ConfigParser .get() calls, which expect a str.

Connecting with FreeTAKServer over TLS

I have exported the .p12 files generated from FTS to .pem format and applying these to the config. I want to check that this is only for the client.p12 and not the server.p12 in the zip file?

When I do so, and disable certificate verification I still get an error on the socket.

2022-09-24 23:56:30,373 pytak INFO pytak.classes.run:281 - Complete: <Task finished name='Task-2' coro=<RclpyTakReceiver.run() done, defined at /home/peter-laptop/spurt/install/lib/python3.8/site-packages/spurt_tak_bridge/tak_bridge.py:59> exception=IncompleteReadError('0 bytes read on a total of undefined expected bytes')>
ERROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-2' coro=<RclpyTakReceiver.run() done, defined at /home/peter-laptop/spurt/install/lib/python3.8/site-packages/spurt_tak_bridge/tak_bridge.py:59> exception=IncompleteReadError('0 bytes read on a total of undefined expected bytes')>
Traceback (most recent call last):
  File "/home/peter-laptop/spurt/install/lib/python3.8/site-packages/spurt_tak_bridge/tak_bridge.py", line 64, in run
    cot = await self.readcot() #reader.readuntil(separator=b'</event>')
  File "/usr/local/lib/python3.8/dist-packages/pytak/classes.py", line 163, in readcot
    return await self.reader.readuntil("</event>".encode("UTF-8"))
  File "/usr/lib/python3.8/asyncio/streams.py", line 629, in readuntil
    raise exceptions.IncompleteReadError(chunk, None)
asyncio.exceptions.IncompleteReadError: 0 bytes read on a total of undefined expected bytes
2022-09-24 23:56:30,384 pytak DEBUG pytak.classes.handle_data:130 - TX: b'<event version="2.0" type="t-x-d-d" uid="takPong" how="m-g" time="2022-09-24T13:56:30.373072Z" start="2022-09-24T13:56:30.373095Z" stale="2022-09-24T14:56:30.373101Z" />'
ERROR:asyncio:Fatal error on SSL transport
protocol: <asyncio.sslproto.SSLProtocol object at 0x7fb13451a520>
transport: <_SelectorSocketTransport closing fd=14>
Traceback (most recent call last):
  File "/usr/lib/python3.8/asyncio/selector_events.py", line 910, in write
    n = self._sock.send(data)
OSError: [Errno 9] Bad file descriptor

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.8/asyncio/sslproto.py", line 685, in _process_write_backlog
    self._transport.write(chunk)
  File "/usr/lib/python3.8/asyncio/selector_events.py", line 916, in write
    self._fatal_error(exc, 'Fatal write error on socket transport')
  File "/usr/lib/python3.8/asyncio/selector_events.py", line 711, in _fatal_error
    self._force_close(exc)
  File "/usr/lib/python3.8/asyncio/selector_events.py", line 723, in _force_close
    self._loop.call_soon(self._call_connection_lost, exc)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 719, in call_soon
    self._check_closed()
  File "/usr/lib/python3.8/asyncio/base_events.py", line 508, in _check_closed
    raise RuntimeError('Event loop is closed``')
RuntimeError: Event loop is closed

This code works with TAKy on a TCP connection. I assume that both TAKy and FTS do not support a mixture of TLS and TCP connections. There is a passphrase associated with the user not sure if this is an issue. I didn't have a problem when connecting iTAK to it.

Add PEM Key passphrase removal instructions

PEM pass phrase: Prompt

The PEM pass phrase: prompt can be the result of using an encrypted (password-protected) PYTAK_TLS_CLIENT_KEY file. Depending on the security requirements in your operating environment,

is the result of using an encrypted (password-protected) PYTAK_TLS_CLIENT_KEY, in your case /etc/pytak.key. You're correct that it's the default TAK Server CoreConfig password. You can avoid the PEM pass phrase: prompt by removing the password protection from the PYTAK_TLS_CLIENT_KEY using the instructions here:

openssl rsa -in pytak.key -out pytak.nopass.key

TypeError: can't multiply sequence by non-int of type 'float'

File "\site-packages\pytak\classes.py", line 99, in fts_compat
pytak_sleep or (pytak.DEFAULT_SLEEP * random.random())

Support Connectivity to ArgusTak

Hi team, I'm trying to use PyTak to connect to ArgusTak, an online Tak Server provider, for the purposes of reading and sending COT messages.

However I am experiencing issues connecting to their service via SSL (they do not support TCP or other protocols) below is my configuration.

I am able to connect via ATAK and WinTak for a point of reference. They require Client Certificate and Password, and TrustStore Certificate and password as per their instructions on their site: https://argustak.com/help/secure-pipe

config["mycottool"] = {
        "COT_URL": "ssl://argustak.com:8089",
        "PYTAK_TLS_CLIENT_CERT": "C:/../client.p12",
        "PYTAK_TLS_CLIENT_PASSWORD": "sample",
        "PYTAK_TLS_CLIENT_CAFILE": "C:/../trust_store_cert.p12",
        "PYTAK_TLS_CLIENT_KEY": "sample",    # Is this the password for the CA file?
        # "PYTAK_TLS_DONT_VERIFY": "1",
        # "PYTAK_TLS_DONT_CHECK_HOSTNAME": "1",
        "FTS_COMPAT": "1",
        "DEBUG": "1"
    }

TypeError: sequence item 0: expected a bytes-like object, str found

Getting the following error on Linux, but not Mac OS. Python 3.9.13:

    b"\n".join([pytak.DEFAULT_XML_DECLARATION, ET.tostring(cot)]) if cot else None
TypeError: sequence item 0: expected a bytes-like object, str found

PYTAK_TLS_CLIENT_KEY in Docs is actually in PYTAK_TLS_CLIENT_PASSWORD

PYTAK_TLS_CLIENT_KEY in Docs is actually in PYTAK_TLS_CLIENT_PASSWORD. I assume the Docs are out of date, but just wanted to point this one out. I followed the basic installation instructions for Pytak shown in the Docs.

Add the ability to import data packages to pytak.

Since most TAK clients (ATAK, WinTAK, TAK Tracker, et al) support .pref/data-packages for server configuration, add this capability to PyTAK as well.

Cryptography functions deprecated

No python-cryptography version for installing is specified, but backends function that crypto uses has been deprecated 3 years and dozens of versions ago
https://cryptography.io/en/latest/changelog/#v36-0-0

Network is unreachable

I'm using adsbcot with adsbexchange pi image. I was previously using DHCP to get an IP but recently moved to a static IP by editing dhcpcd.conf. At that point my Pi stopped outputting COT entirely. I traced the error back to adsbcot exiting out with this error.

After troubleshooting, I believe this is because there is no default route set on this Pi (LAN only). I'm using a very simple config, just output COT to default multicast SA address. Pytak may be thinkin there is no route to the multicast address, although one shouldn't be needed.

[adsbcot]
DUMP1090_URL = http://127.0.0.1/tar1090/data/aircraft.json
POLL_INTERVAL = 10

I resolved this by adding an (unnecessary) default gateway to eth0

pi@adsbexchange:~ $  DEBUG=1 /home/pi/.local/bin/adsbcot -c /etc/adsbcot.ini
Showing Config: %s /etc/adsbcot.ini
==========
<Section: adsbcot>
==========
DEBUG:asyncio:Using selector: EpollSelector
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>
Traceback (most recent call last):
  File "/home/pi/.local/bin/adsbcot", line 8, in <module>
    sys.exit(main())
  File "/home/pi/.local/lib/python3.9/site-packages/adsbcot/commands.py", line 31, in main
    pytak.cli(__name__.split(".")[0])
  File "/home/pi/.local/lib/python3.9/site-packages/pytak/client_functions.py", line 317, in cli
    asyncio.run(main(app_name, config), debug=debug)
  File "/usr/lib/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.9/asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/home/pi/.local/lib/python3.9/site-packages/pytak/client_functions.py", line 262, in main
    await clitool.setup()
  File "/home/pi/.local/lib/python3.9/site-packages/pytak/classes.py", line 237, in setup
    reader, writer = await pytak.protocol_factory(self.config)
  File "/home/pi/.local/lib/python3.9/site-packages/pytak/client_functions.py", line 202, in protocol_factory
    writer = await pytak.create_udp_client(cot_url)
  File "/home/pi/.local/lib/python3.9/site-packages/pytak/client_functions.py", line 69, in create_udp_client
    stream: DatagramClient = await dgconnect((host, port))
  File "/home/pi/.local/lib/python3.9/site-packages/pytak/asyncio_dgram/aio.py", line 269, in connect
    transport, protocol = await loop.create_datagram_endpoint(
  File "/usr/lib/python3.9/asyncio/base_events.py", line 1363, in create_datagram_endpoint
    raise exceptions[0]
  File "/usr/lib/python3.9/asyncio/base_events.py", line 1350, in create_datagram_endpoint
    await self.sock_connect(sock, remote_address)
  File "/usr/lib/python3.9/asyncio/selector_events.py", line 502, in sock_connect
    return await fut
  File "/usr/lib/python3.9/asyncio/selector_events.py", line 507, in _sock_connect
    sock.connect(address)
OSError: [Errno 101] Network is unreachable

deb uses unknown compression of member 'control.tar.zst', giving up

Got this while installing pytak on stratux 1.6r1-eu028-us

Python 3.6: AttributeError: module 'asyncio' has no attribute 'exceptions'

pytak fails to load certs generated by the FTS Web UI

ssl.SSLError: [SSL] PEM lib (_ssl.c:4043)

Reformat RTFD docs as Markdown

Currently the nightmare of RST, rewrite as MD.

Disabling "hello_event"

Is there a simple way to disable the "hello_event" from setting a variable or similar please.

Currently I am commenting out the various functions in the main code which may get overwritten on an update.

Many thanks.

Add support for Multicast receive

Apparently I never completed the multicast portion of PyTAK, namely, it supports multicast send, but not multicast receive.

Potentially uncaught exception for "ConnectionResetError: [Errno 104] Connection reset by peer" in "classes.py"

Has happened that the TAK server has closed the connection . During this process an exception is generated in 'classes.py' with the following traceback:

2023-08-13 21:51:03,268 pytak INFO - Complete: <Task finished name='Task-2' coro=<RXWorker.run() done, defined at /usr/local/lib/python3.11/site-packages/pytak/classes.py:173> exception=ConnectionResetError(104, 'Connection reset by peer')>
08-13-2023 21:51:03.268Z | asyncio | ERROR:Task exception was never retrieved
future: <Task finished name='Task-2' coro=<RXWorker.run() done, defined at /usr/local/lib/python3.11/site-packages/pytak/classes.py:173> exception=ConnectionResetError(104, 'Connection reset by peer')>
Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/pytak/classes.py", line 179, in run
data: bytes = await self.readcot()
^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/pytak/classes.py", line 168, in readcot
return await self.reader.readuntil("".encode("UTF-8"))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/asyncio/streams.py", line 637, in readuntil
await self._wait_for_data('readuntil')
File "/usr/local/lib/python3.11/asyncio/streams.py", line 522, in _wait_for_data
await self._waiter
File "/usr/local/lib/python3.11/asyncio/selector_events.py", line 970, in _read_ready__get_buffer
nbytes = self._sock.recv_into(buf)
^^^^^^^^^^^^^^^^^^^^^^^^^
ConnectionResetError: [Errno 104] Connection reset by peer

TLS Cipher Suite is restricted to FIPS only

You may encounter the following bug when running a pytak based tool with TLS enabled:

ssl.SSLError: [SSL: SSLV3_ALERT_HANDSHAKE_FAILURE] sslv3 alert handshake failure (_ssl.c:1056)

This is possibly due to a TLS Cipher Suite mismatch between the client (in this case, pytak or a pytak-based tool) and the server.

A work-around is available by allowing all Cipher Suites by setting the following environment variable:

export PYTAK_TLS_CLIENT_CIPHERS=ALL

You can enable this inline by pre-pending your command, for example:

PYTAK_TLS_CLIENT_CIPHERS=ALL adsbxcot ...

Or, if you're using a systemd config:

Environment=PYTAK_TLS_CLIENT_CIPHERS=ALL

For reference, the default Cipher Suite is:

DEFAULT_FIPS_CIPHERS: str = "ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES256-GCM-SHA384"

See: https://github.com/ampledata/pytak/blob/main/pytak/constants.py#L32

Update documentation

Add support for authenticated TAK Server connections

Add support for TAK Server connections that require authentication (username & password).

add multicast support for destination

https://github.com/ampledata/pytak/blob/e8a1478c88c0b9597933e30f78a56b78bcb6dfeb/pytak/functions.py#L132

Please consider adding UDP multicast support for CoT destination URL/IP address.

CoT Time/Start/Stale timestamps aren't actually ISO-8601

CoT uses a subset of ISO-8601 selected by W3C in the XML Schema dateTime primitive: https://www.w3.org/TR/xmlschema-2/#dateTime

Technically PyTAK is formatting these timestamps correctly, we're just using the wrong nomenclature to describe the format.

Remove FTS anti-DoS work-around default behavior

PYTAK Currently adds a random sleep after transmitting a CoT to work-around an anti-DoS feature in FTS. Because this is the default behavior all CoT destinations suffer from laggy or bursty output from clients that use PYTAK. The solution is to change the FTS anti-DoS behavior to be optional at run-time, as opposed to being the default.

See: https://github.com/ampledata/pytak/blob/main/pytak/classes.py#L86

Rx Worker Setup

Firstly i want to congratulate the developers on this fantastic project.

I am new at any concept related to TAK and i am still learning how CoT messages work, how they are sent and received. I would like to implement an RX worker that reads CoTs from a TAK server. I understand very well how the TX worker is implemented in this example https://pytak.readthedocs.io/en/latest/examples.html .

Similarly i want to write the script for the RX worker!

Please, can someone help me with some good suggestions?

Thank you!

Add support for flow-tags

Add support for flow-tags in the Event.
The flow-tags element holds "fingerprints" of the system that have processed a particular CoT event instance. This information aids in the routing of CoT messages along a processing chain.

Add constrained logging for systemd invocation.

xml.etree.ElementTree.ParseError: not well-formed

Getting the following for TAK_PROTO=1

2023-12-06 15:14:33,887 pytak INFO - Run: <class 'pytak.classes.CLITool'>
2023-12-06 15:14:33,888 pytak INFO - Run: <class 'pytak.classes.RXWorker'>
2023-12-06 15:14:33,889 pytak INFO - Run: <class 'pytak.classes.TXWorker'>
2023-12-06 15:14:33,989 pytak INFO - Complete: <Task finished name='Task-3' coro=<Worker.run() done, defined at /Users/gba/src/SNS/pytak/pytak/classes.py:108> exception=ParseError('not well-formed (invalid token): line 2, column 268')>
Task exception was never retrieved
future: <Task finished name='Task-3' coro=<Worker.run() done, defined at /Users/gba/src/SNS/pytak/pytak/classes.py:108> exception=ParseError('not well-formed (invalid token): line 2, column 268')>
Traceback (most recent call last):
  File "/Users/gba/src/SNS/pytak/pytak/classes.py", line 127, in run
    await self.handle_data(data)
  File "/Users/gba/src/SNS/pytak/pytak/classes.py", line 155, in handle_data
    await self.send_data(data)
  File "/Users/gba/src/SNS/pytak/pytak/classes.py", line 174, in send_data
    data = takproto.xml2proto(data, proto)
  File "/Users/gba/src/SNS/takproto/takproto/functions.py", line 81, in xml2proto
    event = ET.fromstring(xml)
  File "/Users/gba/.pyenv/versions/3.9.4/lib/python3.9/xml/etree/ElementTree.py", line 1347, in XML
    parser.feed(text)
xml.etree.ElementTree.ParseError: not well-formed (invalid token): line 2, column 268```

Nonexistant Civ icons for certain categories

A General Atomics drone recently flew within ADSB range. ICAO A3C1BE. This is within the Civ ICAO range. However, there is no valid icon type for a civilian affil drone. The affil must be set to mil to have a valid drone icon. The current code results in the standard civ fixed wing icon. Possible solution:

cot_type = f"a-{attitude}-A-M-F-Q"

Either way, Q needs to be capitalized as lowercase Q was not recognized by WinTAK.

PyTAK utils are not producing valid "TAK Protocol Version 0" CoT

Namely, it's missing the <?xml header.

Add 'troubleshooting with netcat' documentation

https://www.youtube.com/watch?v=W2-phl3tY3o

[Question] Connect to TAK Server with Certs and user&password

Hi!

Thanks for that great library !! :)

My question is about the possibility of connecting to the TAK Server (tak.gov) with custom Certs (TLS) & User & Password - is possible with your library?

I did not find anything that could solve my issue :)

Example configuration:

I really appreciate any help you can provide :)

Regards
Przemek

Decouple client cot object from transmission tak_proto version

As a developer, I'd like to build COT event objects as either protobuf or XML and have pytak convert as necessary to the desired tak_protocol bytes for transmission.

This allows folks to develop withtakproto but still be able to function in TAK_PROTO=0 environments.
This allows mixing of new code producing protobuf event objects and legacy code with ET events in the same client.
Likewise, it would allow receipt of both TAK_PROTO 0 and 1 messages at the same time (which seems to be how ATAK is functioning).
 
If you'd be amenable to a PR implementing such, I'm thinking along these lines:

• takproto.proto2xml() to complement the existing takproto.xml2proto()
• Have TXWorker determine type of dequeued object (CotEvent or ET) and serialize according to the configured TAK_PROTO option
• Have RXWorker deserialize according to nature of received bytes and convert (if necessary) to the TAK_PROTO option

Symbol doesn't appear in ATAK after first CoT is sent to TAK Server.

A subsequent (second) CoT send to the TAK Server will show up in ATAK. Investigate why the first CoT does not show up in ATAK.

Performance issues with large queues

Add generic CoT generation function.

There's a lot of CoT generation that falls under D-R-Y, so create a generic cot_gen() function (which can also serve as a reference design).

Encrypted TLS Private Keys (Private Keys with Passphrases)

There is currently no mechanism to set a passphrase for a TLS Private Key (that is, a Private Key that is encrypted with a Passphrase) for a pytak-based tool. This means that you cannot run a pytak-based tool in the background or as a service, as the tool has no means of accepting a passphrase for the private key from a user.

It's probably possible to add support for this, but in the mean-time, a work-around is to decrypt the Private Key and use that with the pytak-tool, instead of the encrypted Private Key.

For example, given the encrypted Private key 'my_cert.key.pem', you can decrypt the private key and remove the passphrase with this command:

$ openssl rsa -in my_cert.key.pem -out my_cert-nopass.key.pem

This will result in a new file my_cert-nopass.key.pem that you can then pass to the pytak-based tool:

export PYTAK_TLS_CLIENT_KEY=my_cert-nopass.key.pem

Or inline:

PYTAK_TLS_CLIENT_KEY=my_cert-nopass.key.pem aprscot ...

Or with systemd:

Environment=PYTAK_TLS_CLIENT_KEY=my_cert-nopass.key.pem

Make log output level configurable.

Is there a config setting to remove the WARN stdout from Pytak?

Not currently, no. Filing this issue to track that.

CoT messages not showing in WinTAK

Hello!
First of all thank you for the the library!

So I am having troubles understanding how to connect to WinTAK. Basically for now I just want to test sending GPS data to WinTAK so it shows as a marker of some type. Later I plan to integrate the device with UAS plugin in order to manage waypoint navigation (I hope this is possible).
The thing is, I have slightly modified the example send.py script, to send fake data to my WinTAK (both the script and WinTAK are running on the same PC).
However I do not receive anything at all in WinTAK (I guess there is network configuration problem in WinTAK) and I can not figure out why.

Here is the code:

import asyncio
import xml.etree.ElementTree as ET

from configparser import ConfigParser

import pytak
import takproto

import time
import datetime

class MySerializer(pytak.QueueWorker):
    """
    Defines how you process or generate your Cursor on Target Events.
    From there it adds the CoT Events to a queue for TX to a COT_URL.
    """

    async def handle_data(self, data):
        """Handle pre-CoT data, serialize to CoT Event, then puts on queue."""
        event = data
        resp = await self.put_queue(event)
        print(resp)

    async def run(self, number_of_iterations=-1):
        """Run the loop for processing or generating pre-CoT data."""
        while 1:
            # data = tak_pong()
            # await self.handle_data(data)
            # print("pong data has been sent")
            # await asyncio.sleep(20)
            now = datetime.datetime.utcnow()
            stale = now
            stale += datetime.timedelta(hours=1)
            now_iso = now.isoformat()[:-4] + "Z"
            stale_iso = stale.isoformat()[:-4] + "Z"
            # 2023-12-05T08:46:22.96Z
            # 2023-12-07T12:01:37.681452Z
            print(now_iso)
            print(stale_iso)
            cot = f"""<?xml version='1.0' encoding='utf-16' standalone='yes'?>
<event version='2.0' uid='S-1-5-21-3327203913-228467099-2355216783-1001' type='a-f-G' time='{now_iso}' start='{now_iso}' stale='{stale_iso}' how='h-e'>
    <point lat='43.545231' lon='-5.661920' hae='9999999' ce='9999999' le='9999999' />
    <detail>
        <takv version='4.1.0.231' platform='WinTAK-CIV' os='Microsoft Windows 11 Home' device='HP HP Laptop 15-da0xxx' />
        <contact callsign='FONG' endpoint='' />
        <uid Droid='FONG' />
        <__group name='Cyan' role='Team Member' />
        <status battery='50' />
        <track course='0.00000000' speed='0.00000000' />
    </detail>
</event>
"""
            
            buf = takproto.xml2proto(cot,  takproto.TAKProtoVer.STREAM)
            print(buf)
            await self.handle_data(buf)
            # await asyncio.sleep(20)
            time.sleep(2)

def tak_pong():
    """Generate a simple takPong CoT Event."""
    root = ET.Element("event")
    root.set("version", "2.0")
    root.set("type", "t-x-d-d")
    root.set("uid", "takPong")
    root.set("how", "m-g")
    root.set("time", pytak.cot_time())
    root.set("start", pytak.cot_time())
    root.set("stale", pytak.cot_time(3600))
    return ET.tostring(root)


async def main():
    """Main definition of your program, sets config params and
    adds your serializer to the asyncio task list.
    """
    config = ConfigParser()
    config["mycottool"] = {"COT_URL": "tcp://192.168.99.108:4242"}
    config = config["mycottool"]

    # Initializes worker queues and tasks.
    clitool = pytak.CLITool(config)
    await clitool.setup()

    # Add your serializer to the asyncio task list.
    clitool.add_tasks(set([MySerializer(clitool.tx_queue, config)]))

    # Start all tasks.
    await clitool.run()


if __name__ == "__main__":
    asyncio.run(main())

Also I attach some WinTAK screenshots which may be useful:

Some help is greatly appreciate it. Thanks!!

BR

Add support for .p12 Certs

Currently we only support .pem

Add support for TAK Protocol Version 1 Mesh & Stream

These are two variations of the protobuf format.

Updated MIL-CIV address range constants

Receiving COT from TAK server

They is a great project! Thank you for sharing.
Have you ever managed to read CoT out of the TAK Server?
I am trying to create a simple client application that sends telemetry info (e.g., GPS) and receivs markers/routs as a Cursor on Target (COT) to and from the Free TAK Server.

Thank you